feat(storage): sync driver + daemon wiring + MCP parity + docs (PR4 of 4)#111
Draft
sidd-27 wants to merge 19 commits into
Draft
feat(storage): sync driver + daemon wiring + MCP parity + docs (PR4 of 4)#111sidd-27 wants to merge 19 commits into
sidd-27 wants to merge 19 commits into
Conversation
Implements the pure-logic core of the bubbaloop storage subsystem from docs/superpowers/specs/2026-05-26-bubbaloop-storage-design.md (spec §17 "PR1"), with no Zenoh dependencies and full unit coverage: - integrity: streaming + one-shot SHA-256, hex/prefix8/base64 helpers, verify (mandatory per-chunk integrity, §11) - recording: manifest.json data model (§4.4) with forward-compatible unknown-field preservation and derived lifecycle/upload state - manifest: atomic write_tmp_then_rename save, load, structural validation (§3.4.5) - profile: profile YAML + full v1 validator (§4.3.1), reserved `pipelines:`->v2 gate (§4.3.2), canonical profile_sha256 (§4.3.3) - secrets: opaque zeroize-on-drop Secret (redacted Debug), secrets.toml chmod 0600 (§4.2) - backend: async StorageBackend trait + LocalFs impl with checksum verification and path-traversal guards; deterministic object-key builders (§8) S3Compat backend, discover/sync/reconcile/replay, CLI/MCP surfaces and the dashboard tab are deferred to later PRs per the spec phasing. Adds sha2, zeroize, async-trait deps. 57 unit tests; clippy-clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Enables building/testing on Apple Silicon (the workspace previously declared only linux-aarch64 and linux-64). No dependency changes. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Implements `storage::reconcile` per spec §3.5 — the user-facing healing primitive that diffs the three sources of truth for a recording (local manifest, local chunk files, remote listing) and converges them. - Applies the §3.5.2 diff matrix row-by-row: uploads missing-remote chunks, re-uploads on remote content mismatch, marks `uploaded_at_ns` when the remote already has a chunk, flags local corruption and chunks missing everywhere, and reports (never touches) orphans. - Optional `restore` re-downloads remote-only chunks, verifying SHA-256 on receipt before writing (§8). - Never deletes anything (§3.5.4) — idempotent and safe to re-run. - Backend-agnostic (`&dyn StorageBackend`) and clock-free (caller passes `now_ns`), so it's fully unit-testable against `LocalFs` as a stand-in remote. Re-uploads the manifest at the end (§3.5.3 step 6). 11 new unit tests (68 storage tests total); clippy-clean within storage. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…quence (PR2) Implements the deterministic core of `storage::sync` per spec §3.4 that the daemon's background queue driver is built from: - BackoffSchedule + SharedBackoff: the shared exponential backoff (§3.4.3) — immediate, then 5s→10s→20s→40s→60s clamp, dead-letter after max_elapsed_time. Backoff is shared across jobs and resets on the first success. - classify(): retryable / terminal / pause-sync error mapping (§3.4.3). Adds Forbidden + NoSuchBucket terminal variants to BackendError for the future S3 backend. - upload_one(): the per-chunk upload sequence (§3.4.4) — re-verify local SHA-256, HEAD-before-PUT idempotency (AlreadyPresent / Conflict / PUT), PUT with the checksum header. - pending_jobs() + mark_chunk_uploaded(): the startup-scan enqueue (§3.4.1 trigger 3) and the atomic manifest mutation recording uploaded_at_ns (§3.4.5). - DeadLetterList: the persisted ~/.bubbaloop/storage/dead_letter.json (§3.4.6) that survives restarts and that reconcile flushes; dedup add + remove + atomic save. The async queue driver (bounded VecDeque + semaphore + the in-process MPSC of ChunkFinalized events) is intentionally deferred — it consumes recorder events that land in a later PR. The decision logic it needs lives here. 12 new unit tests (80 storage tests total); clippy-clean within storage. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…PR2)
Wires the data-plane CLI verbs (spec §5) over the StorageBackend trait, so the
whole flow runs end-to-end against the `local` backend without R2:
- `storage::config` — parses the `[storage]` table of ~/.bubbaloop/config.toml
(§4.1) and a backend factory. `local` builds a LocalFs object store; the cloud
backends (r2/s3/gcs/minio) parse but return BackendNotImplemented until the
S3-compatible backend lands. Secrets are never read here (§4.2).
- `cli/storage.rs` — argh subcommands (NOT clap), output via `log` (never
println!):
- `list [--remote] [--machine] [--tag] [--since]` — scan local manifests (or
remote objects) into a summarised table.
- `info <name>` — pretty-print manifest.json.
- `upload <name>` — flush un-uploaded chunks via sync::upload_one (idempotent).
- `download <name> [--to] [--machine]` — pull + SHA-256-verify each chunk.
- `reconcile <name> [--restore]` — drive storage::reconcile.
- `rm <name> [--remote] [--machine]` — delete local (+ remote objects).
- Registered `Command::Storage` in the binary; pure-local so no Zenoh/daemon.
Recording-control verbs (configure/topics/profile/record/replay) and the cloud
backends are later slices. Collection/formatting logic lives in pure helpers
(`collect_local`, `format_table`, `parse_since`) with 13 new unit tests (93
storage tests total). Smoke-tested end-to-end against a temp HOME: list→upload→
list --remote→download→reconcile --restore→info→rm. clippy-clean within storage.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ficiency) Security: - Validate recording names before they become filesystem paths or object keys. New storage::validate_recording_name (charset [A-Za-z0-9._-], no ./.. /NUL) is enforced inside recording_dir/chunks_dir, closing a path traversal where `storage rm '../../x'` would remove_dir_all an arbitrary directory (and info/download/reconcile reads/writes escaping the sandbox). - Enforce canonical chunk filenames in manifest::validate, so a tampered remote manifest can't smuggle a `../evil` chunk.name into a download write; download also derives the on-disk name from (index, sha256) as defense. Correctness: - `--remote` (list/rm) now refuses a non-remote backend instead of silently operating on the local object store (StorageConfig::is_remote). - run_upload only persists + re-uploads the manifest when ≥1 chunk changed, so an all-failed/all-conflict run can't clobber newer remote manifest state. - reconcile's "present remote, absent local" branch now counts the chunk (marked/verified) instead of mutating the manifest with no report signal. - reconcile marks the recording durably `corrupt` (new Recording.corrupt field, surfaced in `list` status) when a chunk is missing both locally and remotely, instead of only emitting a transient error. - reconcile fetches+hashes the full digest when the backend HEAD returns no checksum, instead of trusting the 8-char key prefix (§8 integrity). - SharedBackoff::on_failure re-anchors on a backward clock step so a hopeless job still dead-letters instead of retrying forever (no saturating_sub=0 trap). - `rm` errors when a name matched nothing (local or remote) instead of a silent exit-0 success on a typo with --remote. Efficiency: - run_upload mutates the manifest in memory and saves once, replacing the per-chunk load+save (O(N) full-manifest rewrites). - collect_remote lists the per-machine prefix when --machine is given rather than enumerating the whole bucket. Adds 5 unit tests (name validation, canonical-name rejection, backward-clock dead-letter, is_remote) and tightens reconcile tests; 98 storage tests pass, clippy-clean. Smoke-tested: traversal/typo/--remote paths rejected, valid local flow intact. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Adds create/list/show/rm over storage::profile (§5): - create builds a Profile from flags and validates it against the v1 schema via profile::save (rejects e.g. ring_buffer without window_secs); --force to overwrite; profile names are run through validate_recording_name. - list scans ~/.bubbaloop/profiles/*.yaml with name/sha256/description. - show prints the on-disk YAML after confirming it parses. - rm deletes the profile (errors if absent). Pure-local, no backend/Zenoh. +1 unit test (split_csv); 99 storage tests pass, clippy-clean. Smoke-tested create/list/show/rm + validation + name guard. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- StorageConfig::save/save_to writes the [storage] table into config.toml, merging via toml::Table so unrelated tables (e.g. [agent]) survive; atomic tmp+rename. - `storage configure --backend ... [--endpoint --region --bucket --retention-days-local --disk-quota-gb --local-path --no-auto-upload]` writes the config; `--access-key-id`/`--secret-access-key` (both required together) write secrets.toml via storage::secrets::save_r2 (chmod 0600). Secrets never touch config.toml. Validates backend kind and requires --bucket for cloud backends. Flag-driven/non-interactive (the wizard will call the same handler). +1 config round-trip test; clippy-clean. Smoke-tested: local/r2 config, secrets split + 0600, [agent] preservation, validation paths. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- storage::discover queries `bubbaloop/**/manifest`, decodes the CBOR node manifests, and groups output topics by machine_id → instance → topic with per-topic liveness (live/idle from still_live && ever_fired), mirroring the `bubbaloop dataflow` query pattern. Zenoh I/O (query_nodes) is thin; the grouping/filtering (group) is pure with --machine / --grep filters. - `storage topics [--machine] [--grep] [-z endpoint] [--timeout-secs]` prints the grouped tree. 4 unit tests on the pure grouping (sort order, machine filter, case-insensitive grep dropping empty nodes). The Zenoh path is compile-checked; it needs a live router + nodes to exercise. 103 storage tests pass, clippy-clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Implements StorageBackend over aws-sdk-s3 for R2/AWS/GCS/MinIO (spec §8): - Explicit static Credentials (never the ambient AWS chain), region (auto for R2), explicit endpoint_url, force_path_style for MinIO. TLS via rustls+ring (aws-smithy-http-client) so the build needs no aws-lc/cmake. - Per-chunk x-amz-checksum-sha256 (base64) on PUT for server-side validation; checksum requested on HEAD/GET so reconcile can compare the full digest. - HEAD→None on not-found; ListObjectsV2 paginated; DeleteObject idempotent. - Error mapping to BackendError per §3.4.3: NoSuchBucket→terminal-pause, AccessDenied/Forbidden→terminal, BadDigest→terminal, NoSuchKey→NotFound, everything else→retryable Io. Optional `s3` cargo feature keeps aws-sdk-s3 out of default builds; the config factory builds S3Compat under the feature (reading creds from secrets.toml) and returns BackendNotImplemented without it. So `bubbaloop storage configure --backend r2 ...` + the existing upload/download/reconcile/rm verbs now reach real R2 when built with --features s3. Verified: `cargo check/test/clippy --features s3` clean (rustls-ring, no cmake); default build unchanged (103 storage tests). Functional R2 round-trips need live credentials (CI/minio) — compile + unit-verified here. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…rface Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Implements the in-tree half of spec §17 PR3: the circular MCAP writer and
sliding-window ring buffer (§3.3.5), and MCAP-based replay into Zenoh (§4.5).
storage::ring_buffer
- `RingBuffer`: pure, IO-free sliding window bounded by BOTH a time window
and a byte cap (256 MiB default OOM safety net), FIFO eviction by log_time,
never drops the sole sample. snapshot() is non-destructive; drain() clears.
- `seal_samples`: the circular MCAP writer — materializes a snapshot into
`chunks/chunk-{idx:06}-{prefix8}.mcap` with rosbag2 wire defaults (786 432-byte
chunks, zstd/fast, CRC on, §3.3.7), dual timestamps, §4.5 channel metadata
(zenoh.encoding / zenoh.topic / bubbaloop.schema_name), streaming SHA-256 with
the prefix embedded at finalize (§3.3.6), rolling a new chunk file at the size
target. Writes manifest.json atomically and returns the assembled Recording.
storage::replay
- `read_recording_messages`: recover messages from a recording's MCAP chunks,
preserving topic/encoding (from channel metadata) and both timestamps.
- `plan` (pure): filter by --topics/--exclude (zenoh key-expr intersection) and
--start-time/--end-time trim, order by the selected timeline (publish_time, or
log_time with --use-log-time), rate-scale offsets, apply --remap.
- `ReplaySink` trait + `run_replay` driver: sleeps between messages to honor the
recorded timing, loops with --loop, cancels promptly via CancellationToken.
CLI: `bubbaloop storage replay <name>` with --rate/--loop/--start-time/
--end-time/--topics/--exclude/--remap/--use-log-time, publishing through a
Zenoh-backed sink (Ctrl-C cancels; the only way to stop --loop).
Deps: add `mcap = "0.24"` (§12). 19 new unit tests (eviction, byte cap,
seal→read round-trip with both timestamps, planner trim/filter/remap/rate,
driver ordering + loop-until-cancel). 122 storage tests pass; clippy-clean.
The `mcap-recorder` marketplace-node relocation (the other half of PR3) lives in
the separate bubbaloop-nodes-official repo, not present in this checkout.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Review of d1adc50 surfaced no correctness bugs; these harden replay fidelity, scaling, and shared-mechanism reuse. - Replay back-pressures instead of dropping: ZenohReplaySink now PUTs with CongestionControl::Block (not the put-default Drop), so a slow link/subscriber stalls the (already-paced) replay rather than silently losing samples (§4.5). - Planner pre-parses --topics/--exclude key patterns once instead of re-parsing them per message (was O(messages × patterns) KeyExpr parses in plan()). - run_replay re-checks cancellation after the inter-message sleep, before the publish, so Ctrl-C in that window doesn't emit one extra sample. - Shared atomic write: extract storage::atomic_write (tmp + fsync + rename); manifest::save_atomic and ring_buffer chunk writes both delegate, so the crash-atomicity guarantee lives in one place (§3.4.5). - Warn before replaying a >=1 GiB recording (v1 buffers it fully in RAM). - Document that McapWriteConfig.chunk_size_bytes is an uncompressed-payload threshold (rosbag2 semantics), so on-disk .mcap sizes differ post-zstd. 122 storage tests pass; PR3 + manifest clippy-clean; s3 feature still builds. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Builds the async sync driver on top of the PR2 sync core — the long-deferred piece, blocked until now only because it consumes ChunkFinalized events. It's self-contained and exercised end-to-end against LocalFs as a stand-in remote, so it needs neither the recorder node nor a live R2. - `ChunkFinalized` event + `SyncHandle`/`spawn` for the recorder to notify the driver over an in-process MPSC channel (§3.4.1 trigger 1). - `SyncQueue`: the synchronous, clock-injected policy core — bounded FIFO with dedup + capacity back-pressure (§3.4.2), the shared-backoff retry gate that pauses the whole queue as a unit and resets on first success (§3.4.3), `NoSuchBucket` → pause-entirely, and dead-letter persistence (§3.4.6). - `run_driver`: thin async shell — absorbs events, runs up to max_concurrent uploads per batch via upload_one (§3.4.4), marks each chunk's manifest (§3.4.5), and coalesces one manifest re-upload per recording per batch. Exits when the channel closes and the queue drains, or on cancel. - `scan_pending`/`scan_pending_in`: the startup scan (§3.4.1 trigger 3) that re-enqueues any chunk with no uploaded_at_ns. 5 tests (queue dedup/cap, retryable→dead-letter past max_elapsed, full seal→upload→manifest-marked round-trip, corrupt-chunk dead-letter, startup scan). 127 storage tests pass; clippy-clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
First slice of the §6.1 CLI↔MCP pairing: the read tools an agent needs to inspect storage state before acting (§6.2 calls this out as critical — agents lack the human's dashboard/`cat` context). - mcp/storage_tools.rs: a second `#[tool_router]` block (merged into the server router via `+`) with `storage_config_get`, `storage_list`, `storage_info`, `storage_profile_list`, `storage_profile_show`. Process-local reads that dispatch straight to the storage core (no PlatformOperations indirection needed for filesystem inspection). Secrets never leave secrets.toml. - rbac.rs: all five mapped to Viewer tier (not unknown-defaults-to-Admin, §6.4), with a test that keys off the canonical STORAGE_READ_TOOLS list. 47 MCP integration tests still pass (router merge is non-disruptive); 131 mcp + 127 storage unit tests pass; clippy-clean. Deferred (separate follow-up): the mutating tools (configure, profile create/rm, upload, download, reconcile, rm) need the CLI handlers refactored to return structured results; the Zenoh/long-running tools (topics_list, replay_start/stop/status) need a session registry; record_* needs the out-of-tree recorder node. The blocking §6.1 parity test lands once that surface is complete. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
… (PR4, §3.4.2) Start the §3.4.2 sync queue driver from `daemon::run` when `[storage].auto_upload` is enabled: seed it with a startup scan of un-uploaded chunks, bridge the daemon's `watch` shutdown signal to the driver's `CancellationToken`, and hold the `SyncHandle` for the daemon's lifetime so the event channel stays open (ready for the recorder's `ChunkFinalized` notifications). Gracefully disabled (with a log line) when there is no `[storage]` config, auto_upload is off, or the backend can't be built. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
….1 parity (PR4)
Complete the §6 storage MCP surface so every `storage` CLI verb has an MCP tool
("anything a human can do at the terminal, an agent can do through MCP"):
- mutating: storage_configure (admin), storage_profile_create/rm (operator),
storage_upload/download/reconcile (operator), storage_rm (admin)
- Zenoh: storage_topics_list (viewer) — live fleet topics via node manifests
- replay: storage_replay_start/stop/status — a background, cancellable replay
session tracked in an in-process registry (new server field, Arc-shared across
clones)
To avoid divergence, the CLI handlers are refactored into reusable cores
(apply_configure / apply_profile_create / download_recording / remove_recording)
that both the CLI verb and the MCP tool call; upload is unified behind a new
`storage::sync::upload_pending` helper (per-recording HEAD-before-PUT loop +
coalesced manifest re-upload) returning an `UploadSummary`. RBAC tiers added for
all new tools; a §6.1 parity unit test asserts every CLI verb maps to a
registered MCP tool, plus a write-tool RBAC-tier test.
Registered as a third ToolRouter merged in BubbaLoopMcpServer::new. Read tools'
ok_json/err_text/StorageNameRequest are reused. 127 storage + 134 mcp unit tests
+ 47 integration_mcp pass; clippy-clean; s3 feature still compiles.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…TURE/ROADMAP (PR4, §14) - docs/concepts/storage.md: user-facing model (recording = manifest + MCAP chunks, content-addressed keys, profiles, secrets isolation) + a 7-command local-backend walkthrough + the MCP parity/RBAC table. - docs/concepts/storage-internals.md: manifest schema + forward-compat, streaming SHA-256 verified on upload and download, the §3.4 sync queue (bounded FIFO, shared backoff, NoSuchBucket pause, dead-letter), HEAD-before-PUT idempotency, the §3.5 reconcile diff-matrix, ring-buffer mode, and paced replay. - ARCHITECTURE.md: storage as the third pillar. - ROADMAP.md: fleet storage phase with in-tree pieces done and mcap-recorder node + dashboard Recordings tab remaining. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- storage_download (MCP): drop the caller-supplied `to` destination — an operator-tier agent could otherwise make the daemon write files at an arbitrary path. The MCP tool now always targets the canonical local recording dir; the CLI keeps `--to` (trusted human, own process). - storage_replay_start (MCP): run the synchronous whole-recording MCAP read on tokio's blocking pool (spawn_blocking) instead of an async worker, and port the ≥1 GiB "loads fully into memory" warning from the CLI path. - ReplayRegistry: evict completed/failed sessions on insert so a long-lived daemon can't leak session entries (a looping replay never left Running); session ids now come from a monotonic counter instead of name+now_ns, removing the same-nanosecond collision that could orphan an uncancellable replay. Register the session before spawning so status/stop are authoritative immediately. - upload_pending: log each chunk's error (with transient/terminal classification) and conflict (with object key + "run reconcile") instead of collapsing them into opaque counts; both the CLI verb and storage_upload MCP tool benefit. Also re-upload manifest.json whenever the recording is fully uploaded — not only when a chunk changed — so a prior run that uploaded all chunks but failed the manifest PUT self-heals on retry rather than requiring reconcile. - Drop misleading unconditional "Starting storage sync..." daemon log and the dead Default derives on ConfigureParams/ProfileCreateParams. Adds ReplayRegistry unit tests (unique ids, terminal-session eviction, stop). 127 storage + 136 mcp + 47 integration_mcp pass; clippy-clean; s3 compiles. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This was referenced Jun 18, 2026
Contributor
Author
|
Stacked PR — review in order. Each merges into
Drafts #109–#111 currently show a cumulative diff against the spec branch; each collapses to its own phase automatically once the PR above it merges. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked split of the storage subsystem — part 4 of 4. Builds on PR3. Scoped to spec §17 PR4 (in-tree).
PR4 — sync driver + daemon wiring + MCP parity
ChunkFinalizedevent +SyncHandle/spawn(in-process MPSC),SyncQueuepolicy core (bounded FIFO with dedup + capacity back-pressure, shared-backoff retry gate that pauses the queue as a unit and resets on first success,NoSuchBucket→pause, dead-letter persistence),run_driverasync shell (batched concurrentupload_one, per-chunk manifest mark, coalesced manifest re-upload), andscan_pendingstartup scan. Exercised end-to-end against LocalFs.daemon::start_storage_syncstarts the driver fromrun()when[storage].auto_upload, seeds a startup scan, bridges thewatchshutdown →CancellationToken, holds theSyncHandlefor daemon lifetime. Gracefully disabled (with a log line) when unconfigured.storage_config_get/list/info/profile_list/profile_show, Viewer) plus mutating/Zenoh/replay tools (storage_configureadmin,storage_profile_create/rmop,storage_upload/download/reconcileop,storage_rmadmin,storage_topics_listviewer,storage_replay_start/stop/status). Replay sessions tracked in an in-processReplayRegistry. CLI handlers refactored into shared cores (apply_configure/apply_profile_create/download_recording/remove_recording,upload_pending→UploadSummary) so CLI verb and MCP tool behave identically (§6.1), guarded by a parity test. RBAC tier per tool (§6.4).docs/concepts/storage.md+storage-internals.md, ARCHITECTURE (third pillar), ROADMAP.Verification
127 storage + 136 mcp unit + 47 integration_mcp tests; clippy-clean;
s3compiles. Includes PR4 review fixes (MCPdownloaddrops caller path / canonical dir only, replayspawn_blocking+ bounded collision-freeReplayRegistry, upload error/conflict logging, manifest self-heal on retry).Not in this PR (follow-ups)
The
mcap-recordernode (separate repo, bubbaloop-nodes-official#38), CLIrecord start/stop/flush+storage_record_*MCP tools that drive it, and the dashboard Recordings tab (§7).🤖 Generated with Claude Code