diff --git a/docs/superpowers/plans/2026-07-04-crucible-depot-core.md b/docs/superpowers/plans/2026-07-04-crucible-depot-core.md new file mode 100644 index 0000000..e1bc1f8 --- /dev/null +++ b/docs/superpowers/plans/2026-07-04-crucible-depot-core.md @@ -0,0 +1,330 @@ +# Crucible Depot Core (Increment 1) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Implement `crucible depot {status,push,verify,get,pull}` with local/cdn/bunny backends in sow-tools, then cut sow-assets-manifest over to it and delete the bash it replaces. + +**Architecture:** New stdlib-only `internal/depot` package with its own `Run(args) int` entrypoint (exit contract 0/1/2/70 is richer than `internal/app.Run`'s 0/1, so dispatch special-cases depot instead of routing through `app.Run`). Backends implement a small interface; a sweep engine does parallel range-GET probes with serial re-confirm. Cutover repo work happens in sow-assets-manifest on its own branch. + +**Tech Stack:** Go 1.26, stdlib `net/http`/`crypto/sha256`/`sync`, `gopkg.in/yaml.v3` (already a dep — add nothing to go.mod, vendorHash must not change). + +**Spec:** `docs/superpowers/specs/2026-07-04-crucible-depot-core-design.md` — read it before any task; it is authoritative. + +## Global Constraints + +- **No new Go dependencies.** stdlib + existing `gopkg.in/yaml.v3` only. +- **HEAD is banned** for existence checks. All probes: IPv4, `GET` with `Range: bytes=0-0`, expect `206` (accept any 2xx). A test must fail if HEAD is introduced. +- **Never prompt, never read stdin** for credentials — read path included. Missing key → clear error on stderr, exit `70`. +- **No cache of "what I uploaded"** — presence is always probed against the real target backend. +- `unconfirmed` is a distinct state from `missing`; exit `2` when unconfirmed-only. +- Exit codes: `0` clean, `1` drift (referenced-but-absent), `2` unconfirmed-only, `64` usage, `70` internal/backend error. +- Blob key: `sha256///`; sha must match `^[0-9a-f]{64}$`. +- Env names (unchanged from bash): `DEPOT_CDN_BASE`/`BUNNY_CDN_BASE` (cdn read base, default `https://cdn-a7f3k9.westgate.pw`), `BUNNY_STORAGE_HOST` (**required for bunny, no default** — spec overrides bash), `BUNNY_STORAGE_ZONE` (default `sow-assets-depot`), `BUNNY_STORAGE_READ_PASSWORD` (falls back to `BUNNY_STORAGE_PASSWORD`) for reads/probes, `BUNNY_STORAGE_PASSWORD` for PUT, `DEPOT_PROBE_JOBS` (default 16), `DEPOT_JOBS` (default 16), `DEPOT_CONNECT_TIMEOUT` (default 10s), `DEPOT_PROBE_MAX_TIME` (default 30s), `DEPOT_CONFIRM_RETRIES` (default 3), `DEPOT_CONFIRM_MAX` (default 200, 0 = unlimited). +- Upload: `PUT` with `AccessKey: ` and `Checksum: ` headers; each sha uploaded at most once per run; probe-before-put. +- Every downloaded blob is re-hashed; deleted on mismatch. +- `cdn` backend writes fail closed (error, never fake success). +- Verify with `make vet test` in sow-tools (run inside `nix develop` if go missing from PATH). +- Commit after each task. **Never commit to main.** sow-tools branch: `depot-core-increment-1`. sow-assets-manifest branch: `crucible-depot-cutover`. sow-tools has an unrelated modified spec file on main — leave it out of commits unless the task says otherwise. + +--- + +### Task 1: Package skeleton — config + manifest parsing + +**Files:** +- Create: `internal/depot/config.go`, `internal/depot/config_test.go` +- Create: `internal/depot/manifest.go`, `internal/depot/manifest_test.go` + +**Interfaces (Produces):** + +```go +package depot + +// config.go +type Config struct { + CDNBase string // DEPOT_CDN_BASE || BUNNY_CDN_BASE || default + StorageHost string // BUNNY_STORAGE_HOST, no default + StorageZone string // BUNNY_STORAGE_ZONE, default "sow-assets-depot" + ReadKey string // BUNNY_STORAGE_READ_PASSWORD || BUNNY_STORAGE_PASSWORD + WriteKey string // BUNNY_STORAGE_PASSWORD + ProbeJobs int // DEPOT_PROBE_JOBS, default 16 + Jobs int // DEPOT_JOBS, default 16 + ConnectTimeout time.Duration // DEPOT_CONNECT_TIMEOUT seconds, default 10s + ProbeMaxTime time.Duration // DEPOT_PROBE_MAX_TIME seconds, default 30s + ConfirmRetries int // DEPOT_CONFIRM_RETRIES, default 3 + ConfirmMax int // DEPOT_CONFIRM_MAX, default 200 (0 = unlimited) +} +func LoadConfig(getenv func(string) string) Config + +// manifest.go +// ReferencedSHAs parses every *.yml in dir (schema: assets[].{path,sha256,size,restype,hak}) +// and returns the deduplicated sha set plus per-sha size (for pull). +func ReferencedSHAs(dir string) (map[string]int64, error) +func ValidSHA(s string) bool // ^[0-9a-f]{64}$ +func BlobKey(sha string) string // "sha256/ab/cd/" +``` + +`LoadConfig` takes `getenv` so tests inject env without `t.Setenv` races. Invalid ints/durations in env → fall back to default (do not error). `ReferencedSHAs` errors on: unreadable dir, no `*.yml` files, YAML parse failure, or an entry whose `sha256` fails `ValidSHA` (report file + path). Manifest struct: + +```go +type manifestFile struct { + Assets []struct { + Path string `yaml:"path"` + SHA256 string `yaml:"sha256"` + Size int64 `yaml:"size"` + } `yaml:"assets"` +} +``` + +- [ ] **Step 1:** `git -C sow-tools switch -c depot-core-increment-1` (from main). +- [ ] **Step 2:** Write failing tests: `TestLoadConfigDefaults`, `TestLoadConfigReadKeyFallback` (READ unset → WriteKey value; READ set → its own), `TestLoadConfigBadIntFallsBack`, `TestBlobKey` (`BlobKey("0b1d…")=="sha256/0b/1d/0b1d…"`), `TestValidSHA` (rejects uppercase, short, non-hex), `TestReferencedSHAs` (temp dir with two yml files sharing one sha → dedup; bad sha → error naming the file; empty dir → error). +- [ ] **Step 3:** `go test ./internal/depot/` — verify FAIL (compile error is fine). +- [ ] **Step 4:** Implement `config.go` + `manifest.go` minimally. +- [ ] **Step 5:** `go test ./internal/depot/` PASS; `gofmt -l -w internal/depot`. +- [ ] **Step 6:** Commit: `feat(depot): config resolution and manifest sha parsing`. + +### Task 2: Local backend + +**Files:** +- Create: `internal/depot/backend.go`, `internal/depot/local.go`, `internal/depot/local_test.go` + +**Interfaces (Produces):** + +```go +// backend.go +type ProbeState int +const ( + Present ProbeState = iota + Absent + Unconfirmed +) +type Backend interface { + Name() string + // Probe returns the existence state of one sha. transient=true means a + // retry might change the answer (feeds the serial confirm loop). + Probe(ctx context.Context, sha string) (state ProbeState, transient bool, err error) + // Get fetches sha into dest (temp file + rename), re-hashes, deletes on mismatch. + Get(ctx context.Context, sha, dest string) error + // Put uploads bytes from src for sha. Read-only backends return an error. + Put(ctx context.Context, sha, src string) error +} + +// local.go +type LocalBackend struct{ Root string } // implements Backend; Name()=="local" +``` + +Local semantics: `Probe` = `os.Stat(Root/BlobKey(sha))`, never transient. `Put` = copy to temp file in dest dir then `os.Rename` (MkdirAll parents). `Get` = copy out + re-hash (hash the source bytes while copying via `io.TeeReader` into `sha256.New()`), delete dest and error on mismatch. `err` from Probe only for real I/O errors (permission), not absence. + +- [ ] **Step 1:** Failing tests: `TestLocalRoundTrip` (Put→Probe Present→Get→bytes equal), `TestLocalProbeAbsent`, `TestLocalGetHashMismatch` (hand-write a corrupt blob file at the keyed path, Get returns error and dest does not exist), `TestLocalPutIdempotent`. +- [ ] **Step 2:** Verify FAIL. **Step 3:** Implement. **Step 4:** PASS + gofmt. +- [ ] **Step 5:** Commit: `feat(depot): backend interface and local backend`. + +### Task 3: Remote backends (cdn, bunny) with fake-Bunny tests + +**Files:** +- Create: `internal/depot/remote.go`, `internal/depot/remote_test.go` + +**Interfaces (Consumes):** `Backend`, `ProbeState`, `Config`, `BlobKey`. +**Produces:** + +```go +// NewBackend returns the backend for name ("local"|"cdn"|"bunny"). +// localRoot is used only for "local". Fails closed: "bunny" with empty +// StorageHost or empty ReadKey returns an error (never prompts); +// unknown name returns an error. +func NewBackend(name, localRoot string, cfg Config) (Backend, error) + +type httpBackend struct { ... } // shared by cdn and bunny +``` + +Implementation requirements: + +```go +// IPv4-only transport — REQUIRED, spec bans IPv6/HEAD probe hazards. +transport := &http.Transport{ + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + d := net.Dialer{Timeout: cfg.ConnectTimeout} + return d.DialContext(ctx, "tcp4", addr) + }, + MaxIdleConnsPerHost: cfg.ProbeJobs, +} +client := &http.Client{Transport: transport, Timeout: cfg.ProbeMaxTime} +``` + +- **Probe** (both remotes): `GET /` with header `Range: bytes=0-0`; drain/close body. 2xx → `Present`; 404/410 → `Absent`; anything else (incl. transport error) → `Unconfirmed, transient=true`. **Never http.Head / MethodHead.** +- cdn: base = `cfg.CDNBase`, no auth headers, `Put` returns `errors.New("cdn backend is read-only")`. +- bunny: probe/read against storage `https:////` with `AccessKey: `. `Get`: try CDN base first, on non-2xx fall back to storage URL with ReadKey (mirrors bash `_depot_bunny_get`); re-hash, delete on mismatch. `Put`: probe first, skip if Present; else `PUT` storage URL, headers `AccessKey: `, `Checksum: `, body streamed from src file; non-2xx → error. `Put` with empty WriteKey → error before any I/O. +- Both `Get`s write via temp file + rename and re-hash exactly like local. + +httptest note: `httptest.NewServer` listens on `127.0.0.1`, which the tcp4 dialer reaches — no test shim needed. + +- [ ] **Step 1:** Failing tests against an `httptest` fake Bunny (a handler recording method+path+headers, serving a blob map): + - `TestProbeUsesRangeGetNotHead` — probe a present sha; assert recorded request has `Method=="GET"` and `Range=="bytes=0-0"`; **also** assert the depot source contains no HEAD probe: the handler returns 500 for any HEAD request, and the test asserts zero HEAD requests were received across the whole test run (this is the anti-HEAD regression tripwire from field note 3). + - `TestProbeStates` — 200/206→Present, 404→Absent, 428→Unconfirmed+transient, 503→Unconfirmed+transient. + - `TestBunnyPutSequence` — Put of an absent sha: recorded sequence is [GET range probe, PUT]; PUT has `Checksum` = uppercase sha and `AccessKey` = write key; Put of a present sha: probe only, no PUT. + - `TestReadWriteKeySplit` — probe uses ReadKey, PUT uses WriteKey (distinct values asserted). + - `TestBunnyNoReadKeyFailsClosed` — `NewBackend("bunny",…)` with empty read key returns error **without reading stdin**: replace `os.Stdin` with the read end of an `os.Pipe()` for the test; after the call, write+close and confirm the pipe is undrained (or simply assert error occurs with no stdin interaction because the code path has no stdin reference — grep-based assertion acceptable: test fails if `internal/depot` sources mention `os.Stdin` or `/dev/tty`). + - `TestBunnyNoStorageHostFailsClosed`. + - `TestGetHashMismatchDeletes` — server returns wrong bytes; Get errors, dest removed. + - `TestCDNPutReadOnly`. +- [ ] **Step 2:** Verify FAIL. **Step 3:** Implement `remote.go`. **Step 4:** PASS + gofmt + `go vet ./...`. +- [ ] **Step 5:** Commit: `feat(depot): cdn and bunny HTTP backends with fake-Bunny tests`. + +### Task 4: Sweep engine — parallel probe + serial confirm + +**Files:** +- Create: `internal/depot/sweep.go`, `internal/depot/sweep_test.go` + +**Interfaces (Consumes):** `Backend`, `Config`, `ProbeState`. +**Produces:** + +```go +type SweepResult struct { + Present []string + Missing []string // confirmed absent (404/410 after retries) + Unconfirmed []string // probe budget exhausted, state unknown +} +// Sweep probes every sha against b: one parallel pass (cfg.ProbeJobs workers, +// sync.WaitGroup + buffered-channel work queue), then any non-Present result +// is re-probed serially with linear backoff (attempt t sleeps t seconds, +// cfg.ConfirmRetries attempts) — unless the non-Present count exceeds +// cfg.ConfirmMax (>0), in which case the remainder is reported Unconfirmed +// without serial re-probe (never collapsed into Missing). +// A transient result that stays non-2xx after retries is Unconfirmed; +// a 404/410 is Missing. Backend errors (non-transient) abort with error. +func Sweep(ctx context.Context, b Backend, shas []string, cfg Config, progress io.Writer) (SweepResult, error) +``` + +Sort `shas` before sweeping (deterministic output). Progress: print `probed X/N` roughly every 1000 blobs to `progress` (may be `io.Discard`). For tests, make the backoff sleeper injectable: package-level `var confirmSleep = time.Sleep`, overridden in tests. + +- [ ] **Step 1:** Failing tests using a stub Backend (function fields): + - `TestSweepAllPresent`, `TestSweepMissing` (404 → Missing, no retries beyond first serial confirm). + - `TestSweepThrottledUnconfirmed` — stub always returns Unconfirmed/transient → after retries lands in `Unconfirmed`, never `Missing`; assert `confirmSleep` called with 1s then 2s. + - `TestSweepTransientRecovers` — first probe 428, serial re-probe 206 → Present. + - `TestSweepConfirmMaxCap` — 5 non-present, ConfirmMax=2 → all 5 Unconfirmed, zero serial re-probes. + - `TestSweepParallelBounded` — 100 shas, ProbeJobs=4; stub counts concurrent in-flight probes (atomic), assert max ≤ 4. +- [ ] **Step 2:** FAIL. **Step 3:** Implement. **Step 4:** PASS + gofmt. +- [ ] **Step 5:** Commit: `feat(depot): parallel probe sweep with serial confirm and unconfirmed state`. + +### Task 5: Commands + Run entrypoint + integration test + +**Files:** +- Create: `internal/depot/run.go`, `internal/depot/run_test.go`, `internal/depot/integration_test.go` + +**Interfaces (Consumes):** everything above. +**Produces:** + +```go +// Run executes a depot subcommand. args[0] is the subcommand +// (status|push|verify|get|pull); returns the process exit code. +// Exit: 0 clean, 1 drift, 2 unconfirmed-only, 64 usage, 70 internal. +func Run(args []string, stdout, stderr io.Writer, getenv func(string) string) int +``` + +Flag parsing: one `flag.FlagSet` per subcommand (stdlib; `internal/depot` is a fresh package, the hand-rolled loop in `internal/app` is not a constraint here). Flags per spec: + +``` +status [--manifests DIR] [--source LOCAL_DEPOT] --target bunny|cdn|local +push [--manifests DIR] --source LOCAL_DEPOT --target bunny +verify [--manifests DIR] --target bunny|cdn [--sample N] +get --target cdn|bunny|local +pull [--manifests DIR] --dest DIR --target cdn|bunny +``` + +`--manifests` default `assets`. `--target local` uses `--source` as root for status, and `--target`-side root comes from `DEPOT_DIR` env when target is local (matches bash `DEPOT_DIR`); document in usage string. Missing required flag / unknown subcommand / no args → usage on stderr, exit 64. + +Command semantics (all build shas via `ReferencedSHAs`, backend via `NewBackend`, sweep via `Sweep`): + +- **status**: sweep; print `referenced= present= missing= unconfirmed=` to stdout, then each missing sha on its own line prefixed `missing ` and each unconfirmed prefixed `unconfirmed `. Exit 1 if missing>0; else 2 if unconfirmed>0; else 0. +- **push**: target must be `bunny` (enforce). Sweep; treat missing∪unconfirmed as to-upload (unconfirmed→re-upload is safe/idempotent per spec). For each, read bytes from `--source` local root; a sha absent from source → report and count as failed. Upload with `cfg.Jobs` parallel workers, each sha at most once. Exit 70 on upload error, 1 if any sha was missing from source, else 0. Print `uploaded= failed=`. +- **verify**: sweep (exit 1/2 as status) **plus**, when clean, download `--sample` N (default 3, 0=skip) random blobs (`math/rand` seeded — pick via deterministic-enough `rand.New(rand.NewSource(time.Now().UnixNano()))`) to temp files via `Get` (which re-hashes); any failure → exit 1. +- **get**: `Get(sha, dest)`; validate sha; exit 0/70. +- **pull**: for every referenced sha, dest path `--dest/BlobKey(sha)`. Incremental: if dest file exists, re-hash it; matching → skip, mismatch → re-download. Download absent/mismatched with `cfg.Jobs` workers via `Get`. Print `pulled= present=`; any failed download → exit 70 after finishing the batch (report each failure). Non-zero missing at the source (404) → exit 1 listing them. + +Wire the binary: **Modify** `internal/dispatch/dispatch.go` — in `runBuilder`, before the `Wired` check/`delegateLegacy` routing, special-case: + +```go +if b.Name == "depot" && b.Wired { + return depot.Run(args[1:], os.Stdout, errw, os.Getenv) +} +``` + +(Exact insertion point: the implementer must read `runBuilder` and place this after subcommand resolution is bypassed — depot does its own subcommand parsing, so pass the raw args through, i.e. `args[1:]` where `args[0]=="depot"` at the dispatcher level, or the full arg slice in `RunBuilder("depot", …)` context. Verify both `crucible depot status …` and `crucible-depot status …` paths hit `depot.Run`.) + +- [ ] **Step 1:** Failing unit tests (`run_test.go`): `TestRunUsage` (no args→64, unknown cmd→64, push --target cdn→64), `TestStatusExitCodes` (drift→1, unconfirmed-only→2 using a bunny backend pointed at an httptest server that 428s), `TestGetInvalidSHA` (→64 or 70, pick 64), `TestNoKeyStatusBunnyFailsClosedNoStdin` (getenv returns empty keys → exit 70, stderr mentions the env var name). +- [ ] **Step 2:** Failing integration test (`integration_test.go`), pure local→local in temp dirs: build manifests yml referencing 3 blobs; source depot has all 3; target depot has 1. Then: `status --target local` → exit 1, missing=2 → `push` (local→local: for the test, allow `push --target local` **only** via an unexported hook? No — keep the spec surface: instead run the integration through bunny backend backed by httptest fake-Bunny with an in-memory blob map, which exercises the real push path) → `status` → 0 → `push` again → 0 uploads (idempotent, assert fake recorded no second PUT). Also `pull` into a pre-populated dest: present-and-hash-ok skipped (fake records no GET for it), corrupt pre-existing file re-downloaded. +- [ ] **Step 3:** FAIL. **Step 4:** Implement `run.go`. **Step 5:** `make vet test` PASS + gofmt. +- [ ] **Step 6:** Commit: `feat(depot): status/push/verify/get/pull commands with exit-code contract`. + +### Task 6: Registry wiring + docs + +**Files:** +- Modify: `internal/dispatch/dispatch.go` (depot Registry entry) +- Modify: `internal/dispatch/dispatch_test.go` (if it asserts wired/unwired sets) + +Flip the depot entry: + +```go +{ + Name: "depot", + Bin: "crucible-depot", + Summary: "content-addressed asset depot (local/cdn/bunny)", + Commands: []Command{ + {Name: "status", Summary: "report referenced-vs-present drift against a backend", Usage: "crucible depot status [--manifests DIR] [--source DIR] --target bunny|cdn|local"}, + {Name: "push", Summary: "upload referenced-but-absent blobs from a local depot", Usage: "crucible depot push [--manifests DIR] --source DIR --target bunny"}, + {Name: "verify", Summary: "existence sweep plus sampled download re-hash", Usage: "crucible depot verify [--manifests DIR] --target bunny|cdn [--sample N]"}, + {Name: "get", Summary: "fetch one blob with sha re-verify", Usage: "crucible depot get --target cdn|bunny|local"}, + {Name: "pull", Summary: "incremental verified pull of every referenced blob", Usage: "crucible depot pull [--manifests DIR] --dest DIR --target cdn|bunny"}, + }, + Wired: true, +}, +``` + +The stale "(SeaweedFS)" wording must not survive anywhere: `grep -ri seaweedfs` across the repo and fix hits. + +- [ ] **Step 1:** Update registry; run `go test ./internal/dispatch/` — fix any test that pinned depot as unwired; add/extend a test asserting `crucible depot` routes to `depot.Run` (e.g. `RunBuilder("depot", []string{"status"})` with no manifests dir returns 70/64, not the unwired 70 message — assert on stderr text). +- [ ] **Step 2:** `make check` (vet+test+shellcheck+yamllint) PASS. Also `make build && ./bin/crucible depot` prints depot usage; `./bin/crucible` interactive menu shows depot entries (manual eyeball via `printf 'q\n' | ./bin/crucible` or equivalent). +- [ ] **Step 3:** Commit: `feat(depot): wire depot into dispatch registry and menu`. + +### Task 7: sow-assets-manifest cutover + +**Repo:** `/home/vicky/Projects/westgate/repositories/migration/sow-assets-manifest`, new branch `crucible-depot-cutover` from main. + +**Files:** +- Delete: `scripts/mirror.sh` +- Modify: `scripts/build-haks.sh` (the `--full` depot-verify block, ~L138-192), `Makefile` (`mirror` target, `haks` target), `.gitea/workflows/release.yml`, `.gitea/workflows/publish-haks.yml`, `.gitea/workflows/build-haks.yml`, `scripts/lib.sh` (header comments only), `AGENTS.md`/`README.md` (author-facing docs) + +**Consumes:** the `crucible depot` CLI from Tasks 1–6. NOTE: CI resolves crucible via the flake input tracking sow-tools' default branch — the sow-tools branch must be merged before this repo's CI can pass. Local verification uses a locally built binary (`sow-tools/bin/crucible` on PATH). Do not run `nix flake update` here; note in the PR description that a `nix flake update sow-tools` (or lock bump) must land with/after the sow-tools merge. + +Changes: + +1. **`release.yml`** "Warm depot cache (verified full pull)" step body (keep name, job env unchanged; drop the ticker/compgen bash): +```yaml + - name: Warm depot cache (verified full pull) + env: + BUNNY_STORAGE_READ_PASSWORD: ${{ secrets.BUNNY_STORAGE_READ_PASSWORD }} + BUNNY_STORAGE_HOST: storage.bunnycdn.com + BUNNY_STORAGE_ZONE: sow-assets-depot + run: | + nix develop --command bash -c ' + set -euo pipefail + crucible depot status --manifests assets --target bunny + crucible depot pull --manifests assets --dest "$DEPOT_CACHE_DIR" --target cdn + ' +``` + (If the `BUNNY_STORAGE_READ_PASSWORD` secret does not exist in the repo, use `secrets.BUNNY_STORAGE_PASSWORD` and say so in the PR body. Check what other steps use.) +2. **`publish-haks.yml`** — same swap of the identical step. +3. **`build-haks.yml`** — the PR-check step keeps `build-haks.sh --dry-run --no-verify --base "$base"` (structural check) and **adds** `crucible depot verify --manifests assets --target cdn --sample 3` as the cdn-backend verification. +4. **`scripts/build-haks.sh`** — delete the depot existence-sweep + sample-download block inside `--dry-run` (the `depot_has_many`/`depot_get` verify at ~L138-185); replace with a comment pointing at `crucible depot verify`. Remove now-unused locals. `shellcheck scripts/build-haks.sh` must pass. +5. **`Makefile`** — `mirror:` target now runs `crucible depot pull --manifests assets --dest "$(DEPOT_DIR)" --target cdn`. `haks:` gains a trailing non-fatal drift report: `-crucible depot status --manifests assets --target cdn` (leading `-` so drift never fails a local build) and an echo pointing at `crucible depot push --source workspace/depot --target bunny` as the standard post-edit step. +6. **`scripts/lib.sh`** — no function deletions (caller analysis: every depot function used by mirror.sh/build-haks.sh retains Increment 2–3 callers). Add a header comment block at the top of the depot section: `# NOTE: this bash depot layer is being retired. Increment 1 moved mirror/verify to 'crucible depot' (sow-tools). import/sync go in Increment 2; export/pack/publish/artifact/cdn_purge in Increment 3. Do not add new callers.` **Do not touch the backend-keyed stat-cache in import.sh** (spec forbids it). +7. **Docs** — in README.md (or AGENTS.md where depot workflow is documented): document `crucible depot status --manifests assets --target cdn` as the canonical credential-free "is my content live?" author check, and `crucible depot push --source workspace/depot --target bunny` as the standard post-edit publish step. + +- [ ] **Step 1:** Branch; make all edits above. +- [ ] **Step 2:** Verify: `shellcheck scripts/*.sh`; `yamllint .gitea`; `grep -rn "mirror.sh" . --exclude-dir=.git` returns nothing; with sow-tools' `bin` on PATH run `crucible depot status --manifests assets --target cdn` for real (credential-free) and confirm it produces a sane report (exit 0 or 2 acceptable; exit 1 means real drift — report it, don't hide it). +- [ ] **Step 3:** Commit: `feat: cut depot mirror/verify over to crucible depot, retire mirror.sh`. + +## Self-Review Notes + +- Spec coverage: command surface (T5), drift model + unconfirmed/exit 2 (T4/T5), backends + env + no-prompt (T1/T3), IPv4 range probe + HEAD ban test (T3), Checksum PUT (T3), pull incremental (T5), cutover of all three workflows + Makefile + lib.sh headers + author docs (T7), registry/SeaweedFS fix + menu (T6), testing section requirements all mapped (T1–T5), interim contract `make haks` drift report (T7.5). +- Deliberate deviations from spec, justified: `status --target local` root via `DEPOT_DIR`; push treats source-missing blobs as exit 1; pull download failure exits 70 (backend error) but source-404 exits 1 (drift). +- Out of scope, per spec non-goals: import/sync, prune/gc/export, mdl checks, rich TUI, cdn_purge. diff --git a/docs/superpowers/specs/2026-07-04-crucible-depot-core-design.md b/docs/superpowers/specs/2026-07-04-crucible-depot-core-design.md index ec13993..bb9b4bd 100644 --- a/docs/superpowers/specs/2026-07-04-crucible-depot-core-design.md +++ b/docs/superpowers/specs/2026-07-04-crucible-depot-core-design.md @@ -124,9 +124,10 @@ Increment 1. This spec covers **Increment 1** only. Roadmap for context: -- **Increment 1 (this spec):** `crucible depot {status, push, verify, get}` + - blob backends (`local` / `cdn` / `bunny`). Cut over `mirror`/verify/release - gate. Delete the corresponding `lib.sh` depot layer. +- **Increment 1 (this spec):** `crucible depot {status, push, verify, get, pull}` + + blob backends (`local` / `cdn` / `bunny`). Cut over `mirror`/verify/release + gate. Delete the `lib.sh` depot functions that lose their last caller (see + Cutover for what actually remains until Increments 2–3). - **Increment 2:** `crucible depot import` / `sync` (edit-tree → manifest+depot), replacing `import.sh` / `sync-assets.sh`. Retire the stat-cache concept. - **Increment 3:** mdl integrity checks, `mirror`/`pull` ergonomics, `prune`, @@ -141,8 +142,17 @@ crucible depot status [--manifests DIR] [--source LOCAL_DEPOT] --target bunny|c crucible depot push [--manifests DIR] --source LOCAL_DEPOT --target bunny crucible depot verify [--manifests DIR] --target bunny|cdn [--sample N] crucible depot get --target cdn|bunny|local +crucible depot pull [--manifests DIR] --dest DIR --target cdn|bunny ``` +- `pull` is the bulk fetch that replaces `mirror.sh`: every referenced sha into + `--dest/sha256/ab/cd/`, **incremental** (skip blobs already present with a + matching hash — the release warm step re-runs against a persistent + `/var/cache` dir holding ~69k blobs; a non-incremental pull would re-download + the corpus every release), sha re-verified on download, parallel with the same + probe/transfer bounds as `status`. Without `pull`, Increment 1 cannot delete + `mirror.sh` (single-blob `get` is not a replacement). + - `--manifests` defaults to `assets/` (the repo's manifest dir). Confirmed decision: `crucible depot` parses `assets/*.yml` directly (same coupling model as `crucible-hak` reading hak manifests); the caller does not pre-extract sha @@ -159,8 +169,26 @@ no cache of "what I uploaded." - **`status`** — collect every `sha256` referenced by `--manifests/*.yml`, run one parallel existence sweep against `--target`, report - `{referenced, present, missing}`. Non-zero exit if any referenced blob is - missing. This is the release pre-flight gate. + `{referenced, present, missing, unconfirmed}`. Non-zero exit if any referenced + blob is missing. This is the release pre-flight gate. + + **`unconfirmed` is a distinct state, not `missing`.** Any non-2xx from the + parallel sweep is re-probed serially with backoff (the bash + `depot_confirm_absent` behavior). If the serial re-probe budget is exhausted + (bash caps at 200; keep a cap, `DEPOT_CONFIRM_MAX`), the remainder is reported + as `unconfirmed` — **not** collapsed into `missing`. For `push` the + distinction is harmless (unconfirmed → treat as absent → re-upload; wasteful + but idempotent and safe). For `status` as the release gate it is critical: a + throttled CDN edge must produce "N unconfirmed, retry" (distinct exit code, + proposed `2`), never a false "thousands missing" hard-fail — exactly the + over-reporting the corrected v0.1.5 drift measurement documented above. + + **Target authority split:** `bunny` probes storage — the origin, authoritative, + needs the read key; this is what the release gate uses. `cdn` probes the edge — + credential-free, what players actually fetch, the author-facing "is my content + live?" check; the edge cache can briefly lag storage (bash carries `cdn_purge` + for this reason). A blob `present` on bunny but `missing` on cdn is + propagation lag, not drift. - **`push`** — run `status`, then for each missing sha read the bytes from `--source` and upload to `--target`. Stateless and idempotent; re-running does nothing once clean. **Cannot silently skip an upload** — "should I upload this?" @@ -189,13 +217,25 @@ untouched; flags are local aliases): - `BUNNY_STORAGE_HOST`, `BUNNY_STORAGE_ZONE` — Bunny storage endpoint. - **Read/write key split (preserved):** `BUNNY_STORAGE_READ_PASSWORD` (falls back to `BUNNY_STORAGE_PASSWORD`) for probes/reads; `BUNNY_STORAGE_PASSWORD` for - `PUT`. + `PUT`. Note for the `release.yml` cutover: `status --target bunny` as the gate + step needs at least the read key exported in that step's env. +- **No prompting, read path included.** The no-prompt rule from field note 1 + applies to every credentialed operation, not just `push`: `status`/`verify`/ + `get` against `bunny` with no read key fail closed with a clear message — + never a hang, never a tty read. +- `BUNNY_STORAGE_HOST` stays **env-required, no baked default** — the bash never + defaulted it (`depot_require_write` defaults only the zone + `sow-assets-depot` and the CDN base); resolving the earlier open question. - Concurrency: `DEPOT_PROBE_JOBS` (read probes), `DEPOT_JOBS` (uploads); `DEPOT_CONNECT_TIMEOUT`, `DEPOT_PROBE_MAX_TIME` per-transfer bounds. Invariants carried over exactly: -- **Existence probe** is a 1-byte range request (no body download). +- **Existence probe** is an IPv4 1-byte range request (no body download) — for + **all** backends. This is a deliberate deviation from `lib.sh`, whose `cdn` + backend probes with HEAD (`_depot_cdn_probe_code`); the "faithful port" + clause above does not extend to the probe method. HEAD is banned (see field + note 3 and the test requirement there). - **Upload** is `PUT` with `Checksum: ` so Bunny rejects corrupt writes server-side; each sha uploaded at most once per run. - **Fail-safe:** a blob with no confirmed-present reply is treated as absent and @@ -223,26 +263,68 @@ assets/*.yml ──parse──▶ referenced sha set ─┐ Increment 1 deletes the bash it replaces (no wrappers left behind): -- `scripts/mirror.sh` → `crucible depot get`/pull path; script removed. +- `scripts/mirror.sh` → `crucible depot pull`; script removed. - `build-haks.sh`'s `--full` depot-verify block → `crucible depot verify`. -- `release.yml` step "Warm depot cache" → `crucible depot status` (gate) then the - fetch path; the inline bash ticker/compgen logic goes away. -- `lib.sh` depot layer (`depot_require`, `depot_require_write`, `_depot_*_has`, - `_depot_*_get`, `_depot_*_put*`, `_depot_*_probe*`, `blob_key`, backend - dispatch — ~250+ lines) is removed once no remaining bash references it. - Anything still needed by Increments 2–3 is deleted when those land, not kept as - a compatibility shim. +- **All three depot-touching workflows switch in the same change** (not just + `release.yml`): + - `release.yml` "Warm depot cache (verified full pull)" → + `crucible depot status --target bunny` (gate) then + `crucible depot pull --dest $DEPOT_CACHE_DIR`; the inline bash + ticker/compgen logic goes away. + - `publish-haks.yml` mirrors the same warm+verify+pack flow — same swap. + - `build-haks.yml` (PR check, cdn-backend verification) → + `crucible depot verify --target cdn`. +- Registry wiring: the `depot` entry in `internal/dispatch` flips to + `Wired: true` with its `Commands` list populated (status/push/verify/get/pull) + and the stale "(SeaweedFS)" summary corrected. This automatically puts depot + in the interactive `crucible` menu alongside the other wired tools — no + menu-specific code needed. + +**What `lib.sh` deletion actually looks like.** The depot layer cannot be fully +removed in Increment 1 — it is still referenced by scripts outside this +increment's scope: + +- `import.sh` / `sync-assets.sh` (`make haks`, `make import`, `make sync`) — + Increment 2. +- `export.sh` (`make pull`), `pack-haks.sh` (release-time `DEPOT_BACKEND=bunny` + packing), `publish-release.sh` + `release_put`, and the path-addressed + `artifact_has/get/put` + `cdn_purge` helpers — Increment 3 (these were + previously unlisted; they are now explicitly Increment 3 scope). + +Increment 1 deletes the functions whose last caller it removes (the +mirror/verify probe-and-get paths) and leaves the rest with a header comment +marking them scheduled for deletion in Increments 2–3. No compatibility shims, +no new wrappers — but no pretending `lib.sh` dies before its last caller does. Consumers call `crucible depot …` directly. Nothing new is wrapped. +### Interim contract for local builds (until Increment 2) + +`make haks` keeps running bash `sync-assets.sh` with `DEPOT_BACKEND=local` until +Increment 2 replaces import/sync. The gap this leaves — local imports creating +blobs Bunny never sees — is held closed by two things: + +- The backend-keyed stat-cache + ([`sow-assets-manifest#29`](https://git.westgate.pw/ShadowsOverWestgate/sow-assets-manifest/pulls/29)) + stays exactly as it is; do not "simplify" it (see the 2026-07-04 update above). +- `make haks` gains a trailing `crucible depot status --manifests assets + --target cdn` (credential-free, report-only, non-fatal) so drift is visible + the moment it is created instead of at release time; and + `crucible depot push --source workspace/depot --target bunny` is documented + as the standard post-edit step (field note 4's ask, landed here). Upload when + you want the content live; build local-only when you don't — but either way + the drift is *reported*, never masked. + ## Exit-code contract Matches Crucible's fail-closed convention: - `0` — clean / success. -- distinct non-zero (proposed `1`) — **drift found** (`status`/`push` saw - referenced-but-absent blobs), so `release.yml` can gate on `status` before - packing. +- `1` — **drift found** (`status`/`push` saw referenced-but-absent blobs), so + `release.yml` can gate on `status` before packing. +- `2` — **unconfirmed only**: no confirmed-missing blobs, but the probe budget + ran out before every blob got a confirmed reply (throttled edge). The gate + should retry, not report drift. - `70` — internal/backend error; never a faked result. ## Testing @@ -255,6 +337,10 @@ Matches Crucible's fail-closed convention: `Checksum: ` header, and the read/write key split — no real Bunny. - **Integration (local→local)**: `status` shows drift → `push` clears it → `status` clean → `push` again is a no-op (idempotency). +- **New-surface tests**: fake-Bunny throttling (`428`) → blobs land in + `unconfirmed`, exit `2`, never `missing`; `pull` into a pre-populated dest + skips present-and-hash-ok blobs (incremental); credentialed op with no key + exits non-zero without reading stdin (read path and write path both). ## Non-goals (Increment 1) @@ -262,7 +348,11 @@ Matches Crucible's fail-closed convention: - mdl integrity checks, `prune`, `gc`, `export`, `mirror` ergonomics (Increment 3). - Any change to the depot wire protocol, blob layout, or Bunny account config. -- A TUI. +- A depot-specific rich TUI. **Direction note:** Crucible is headed toward being + a TUI overall — the existing interactive dispatcher menu is the first step, + and depot joins it for free via the registry wiring above. What Increment 1 + does not build is anything richer than that menu (progress UIs, pickers, + dashboards); those come with the Crucible-wide TUI work, not per-tool. ## Immediate operational payoff @@ -276,8 +366,8 @@ Increment 1 first both fixes the tooling and clears the current outage. ## Risks / open questions -- `BUNNY_STORAGE_HOST` default: confirm the canonical storage host to bake as the - fallback (bash referenced it without a visible default in the reviewed range). +- ~~`BUNNY_STORAGE_HOST` default~~ — resolved: env-required, no baked default + (see Backends & config). - Manifest schema coupling: `crucible depot` now depends on the `assets/*.yml` shape (`assets[].{path,sha256,size,restype,hak}`). Acceptable and intentional, but a schema change now touches Crucible. diff --git a/internal/depot/backend.go b/internal/depot/backend.go new file mode 100644 index 0000000..6e66f6e --- /dev/null +++ b/internal/depot/backend.go @@ -0,0 +1,24 @@ +package depot + +import "context" + +type ProbeState int + +const ( + Present ProbeState = iota + Absent + Unconfirmed +) + +// Backend defines the interface for blob storage backends. +type Backend interface { + // Name returns the backend name. + Name() string + // Probe returns the existence state of one sha. transient=true means a + // retry might change the answer (feeds the serial confirm loop). + Probe(ctx context.Context, sha string) (state ProbeState, transient bool, err error) + // Get fetches sha into dest (temp file + rename), re-hashes, deletes on mismatch. + Get(ctx context.Context, sha, dest string) error + // Put uploads bytes from src for sha. Read-only backends return an error. + Put(ctx context.Context, sha, src string) error +} diff --git a/internal/depot/config.go b/internal/depot/config.go new file mode 100644 index 0000000..f9d6d92 --- /dev/null +++ b/internal/depot/config.go @@ -0,0 +1,102 @@ +package depot + +import ( + "strconv" + "time" +) + +type Config struct { + CDNBase string + StorageHost string + StorageZone string + ReadKey string + WriteKey string + ProbeJobs int + Jobs int + ConnectTimeout time.Duration + ProbeMaxTime time.Duration + ConfirmRetries int + ConfirmMax int +} + +func LoadConfig(getenv func(string) string) Config { + cfg := Config{ + CDNBase: "https://cdn-a7f3k9.westgate.pw", + StorageZone: "sow-assets-depot", + ProbeJobs: 16, + Jobs: 16, + ConnectTimeout: 10 * time.Second, + ProbeMaxTime: 30 * time.Second, + ConfirmRetries: 3, + ConfirmMax: 200, + } + + // DEPOT_CDN_BASE || BUNNY_CDN_BASE + if v := getenv("DEPOT_CDN_BASE"); v != "" { + cfg.CDNBase = v + } else if v := getenv("BUNNY_CDN_BASE"); v != "" { + cfg.CDNBase = v + } + + // BUNNY_STORAGE_HOST (no default) + cfg.StorageHost = getenv("BUNNY_STORAGE_HOST") + + // BUNNY_STORAGE_ZONE (default "sow-assets-depot") + if v := getenv("BUNNY_STORAGE_ZONE"); v != "" { + cfg.StorageZone = v + } + + // BUNNY_STORAGE_PASSWORD + cfg.WriteKey = getenv("BUNNY_STORAGE_PASSWORD") + + // BUNNY_STORAGE_READ_PASSWORD || BUNNY_STORAGE_PASSWORD + if v := getenv("BUNNY_STORAGE_READ_PASSWORD"); v != "" { + cfg.ReadKey = v + } else { + cfg.ReadKey = cfg.WriteKey + } + + // DEPOT_PROBE_JOBS (default 16) + if v := getenv("DEPOT_PROBE_JOBS"); v != "" { + if i, err := strconv.Atoi(v); err == nil { + cfg.ProbeJobs = i + } + } + + // DEPOT_JOBS (default 16) + if v := getenv("DEPOT_JOBS"); v != "" { + if i, err := strconv.Atoi(v); err == nil { + cfg.Jobs = i + } + } + + // DEPOT_CONNECT_TIMEOUT (default 10s) + if v := getenv("DEPOT_CONNECT_TIMEOUT"); v != "" { + if d, err := time.ParseDuration(v + "s"); err == nil { + cfg.ConnectTimeout = d + } + } + + // DEPOT_PROBE_MAX_TIME (default 30s) + if v := getenv("DEPOT_PROBE_MAX_TIME"); v != "" { + if d, err := time.ParseDuration(v + "s"); err == nil { + cfg.ProbeMaxTime = d + } + } + + // DEPOT_CONFIRM_RETRIES (default 3) + if v := getenv("DEPOT_CONFIRM_RETRIES"); v != "" { + if i, err := strconv.Atoi(v); err == nil { + cfg.ConfirmRetries = i + } + } + + // DEPOT_CONFIRM_MAX (default 200, 0 = unlimited) + if v := getenv("DEPOT_CONFIRM_MAX"); v != "" { + if i, err := strconv.Atoi(v); err == nil { + cfg.ConfirmMax = i + } + } + + return cfg +} diff --git a/internal/depot/config_test.go b/internal/depot/config_test.go new file mode 100644 index 0000000..e487fab --- /dev/null +++ b/internal/depot/config_test.go @@ -0,0 +1,115 @@ +package depot + +import ( + "testing" + "time" +) + +func TestLoadConfigDefaults(t *testing.T) { + getenv := func(key string) string { + return "" + } + cfg := LoadConfig(getenv) + if cfg.CDNBase != "https://cdn-a7f3k9.westgate.pw" { + t.Errorf("CDNBase: got %q, want %q", cfg.CDNBase, "https://cdn-a7f3k9.westgate.pw") + } + if cfg.StorageZone != "sow-assets-depot" { + t.Errorf("StorageZone: got %q, want %q", cfg.StorageZone, "sow-assets-depot") + } + if cfg.ProbeJobs != 16 { + t.Errorf("ProbeJobs: got %d, want 16", cfg.ProbeJobs) + } + if cfg.Jobs != 16 { + t.Errorf("Jobs: got %d, want 16", cfg.Jobs) + } + if cfg.ConnectTimeout != 10*time.Second { + t.Errorf("ConnectTimeout: got %v, want 10s", cfg.ConnectTimeout) + } + if cfg.ProbeMaxTime != 30*time.Second { + t.Errorf("ProbeMaxTime: got %v, want 30s", cfg.ProbeMaxTime) + } + if cfg.ConfirmRetries != 3 { + t.Errorf("ConfirmRetries: got %d, want 3", cfg.ConfirmRetries) + } + if cfg.ConfirmMax != 200 { + t.Errorf("ConfirmMax: got %d, want 200", cfg.ConfirmMax) + } +} + +func TestLoadConfigReadKeyFallback(t *testing.T) { + // When READ unset, falls back to WriteKey + getenv := func(key string) string { + if key == "BUNNY_STORAGE_PASSWORD" { + return "write-key" + } + return "" + } + cfg := LoadConfig(getenv) + if cfg.ReadKey != "write-key" { + t.Errorf("ReadKey fallback: got %q, want %q", cfg.ReadKey, "write-key") + } + if cfg.WriteKey != "write-key" { + t.Errorf("WriteKey: got %q, want %q", cfg.WriteKey, "write-key") + } + + // When READ is set, uses its own value + getenv = func(key string) string { + if key == "BUNNY_STORAGE_READ_PASSWORD" { + return "read-key" + } + if key == "BUNNY_STORAGE_PASSWORD" { + return "write-key" + } + return "" + } + cfg = LoadConfig(getenv) + if cfg.ReadKey != "read-key" { + t.Errorf("ReadKey explicit: got %q, want %q", cfg.ReadKey, "read-key") + } + if cfg.WriteKey != "write-key" { + t.Errorf("WriteKey: got %q, want %q", cfg.WriteKey, "write-key") + } +} + +func TestLoadConfigBadIntFallsBack(t *testing.T) { + getenv := func(key string) string { + if key == "DEPOT_PROBE_JOBS" { + return "not-a-number" + } + if key == "DEPOT_JOBS" { + return "invalid" + } + if key == "DEPOT_CONNECT_TIMEOUT" { + return "bad" + } + if key == "DEPOT_PROBE_MAX_TIME" { + return "wrong" + } + if key == "DEPOT_CONFIRM_RETRIES" { + return "nope" + } + if key == "DEPOT_CONFIRM_MAX" { + return "nah" + } + return "" + } + cfg := LoadConfig(getenv) + if cfg.ProbeJobs != 16 { + t.Errorf("ProbeJobs fallback: got %d, want 16", cfg.ProbeJobs) + } + if cfg.Jobs != 16 { + t.Errorf("Jobs fallback: got %d, want 16", cfg.Jobs) + } + if cfg.ConnectTimeout != 10*time.Second { + t.Errorf("ConnectTimeout fallback: got %v, want 10s", cfg.ConnectTimeout) + } + if cfg.ProbeMaxTime != 30*time.Second { + t.Errorf("ProbeMaxTime fallback: got %v, want 30s", cfg.ProbeMaxTime) + } + if cfg.ConfirmRetries != 3 { + t.Errorf("ConfirmRetries fallback: got %d, want 3", cfg.ConfirmRetries) + } + if cfg.ConfirmMax != 200 { + t.Errorf("ConfirmMax fallback: got %d, want 200", cfg.ConfirmMax) + } +} diff --git a/internal/depot/integration_test.go b/internal/depot/integration_test.go new file mode 100644 index 0000000..5515a9b --- /dev/null +++ b/internal/depot/integration_test.go @@ -0,0 +1,183 @@ +package depot + +import ( + "bytes" + "fmt" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" +) + +// TestIntegrationPushStatusPull drives status/push/pull end-to-end against a +// fake Bunny backend (real bunny HTTP path, in-memory blob store) since the +// command surface has no local->local push (push --target must be bunny). +func TestIntegrationPushStatusPull(t *testing.T) { + manifestsDir := t.TempDir() + sourceDir := t.TempDir() + + blobs := map[string]string{ + shaOf("blob-one"): "blob-one", + shaOf("blob-two"): "blob-two", + shaOf("blob-three"): "blob-three", + } + var manifest strings.Builder + manifest.WriteString("assets:\n") + for sha, content := range blobs { + manifest.WriteString(fmt.Sprintf(" - path: %s\n sha256: %s\n size: %d\n", sha, sha, len(content))) + blobPath := filepath.Join(sourceDir, BlobKey(sha)) + if err := os.MkdirAll(filepath.Dir(blobPath), 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(blobPath, []byte(content), 0644); err != nil { + t.Fatal(err) + } + } + if err := os.WriteFile(filepath.Join(manifestsDir, "manifest.yml"), []byte(manifest.String()), 0644); err != nil { + t.Fatal(err) + } + + f := newFakeBunny() + // Pre-seed one of the three blobs on the target so status starts with drift=2. + var oneSHA string + for sha := range blobs { + oneSHA = sha + break + } + f.blobs[oneSHA] = []byte(blobs[oneSHA]) + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + env := map[string]string{ + "BUNNY_STORAGE_HOST": hostPort(srv), + "BUNNY_STORAGE_READ_PASSWORD": "readkey", + "BUNNY_STORAGE_PASSWORD": "writekey", + } + getenv := testGetenv(env) + + // status: expect exit 1, missing=2 + var out, errb bytes.Buffer + code := Run([]string{"status", "--manifests", manifestsDir, "--target", "bunny"}, &out, &errb, getenv) + if code != 1 { + t.Fatalf("status: expected 1, got %d (stdout=%s stderr=%s)", code, out.String(), errb.String()) + } + if !bytesContains(out.String(), "missing=2") { + t.Fatalf("status: expected missing=2, got %s", out.String()) + } + + // push: uploads the 2 missing blobs + out.Reset() + errb.Reset() + code = Run([]string{"push", "--manifests", manifestsDir, "--source", sourceDir, "--target", "bunny"}, &out, &errb, getenv) + if code != 0 { + t.Fatalf("push: expected 0, got %d (stdout=%s stderr=%s)", code, out.String(), errb.String()) + } + if !bytesContains(out.String(), "uploaded=2 failed=0") { + t.Fatalf("push: expected uploaded=2 failed=0, got %s", out.String()) + } + + putCountAfterFirstPush := 0 + for _, r := range f.requestsSnapshot() { + if r.Method == "PUT" { + putCountAfterFirstPush++ + } + } + if putCountAfterFirstPush != 2 { + t.Fatalf("expected 2 PUTs after first push, got %d", putCountAfterFirstPush) + } + + // status: now clean + out.Reset() + errb.Reset() + code = Run([]string{"status", "--manifests", manifestsDir, "--target", "bunny"}, &out, &errb, getenv) + if code != 0 { + t.Fatalf("status after push: expected 0, got %d (stdout=%s stderr=%s)", code, out.String(), errb.String()) + } + + // push again: idempotent, no new PUTs, uploaded=0 + out.Reset() + errb.Reset() + code = Run([]string{"push", "--manifests", manifestsDir, "--source", sourceDir, "--target", "bunny"}, &out, &errb, getenv) + if code != 0 { + t.Fatalf("second push: expected 0, got %d (stdout=%s stderr=%s)", code, out.String(), errb.String()) + } + if !bytesContains(out.String(), "uploaded=0 failed=0") { + t.Fatalf("second push: expected uploaded=0 failed=0, got %s", out.String()) + } + putCountAfterSecondPush := 0 + for _, r := range f.requestsSnapshot() { + if r.Method == "PUT" { + putCountAfterSecondPush++ + } + } + if putCountAfterSecondPush != putCountAfterFirstPush { + t.Fatalf("expected no additional PUTs on idempotent push, before=%d after=%d", putCountAfterFirstPush, putCountAfterSecondPush) + } + + // pull into a dest dir: one pre-populated correct file (skipped, no GET), + // one pre-populated corrupt file (re-downloaded), one absent (downloaded). + destDir := t.TempDir() + shas := make([]string, 0, len(blobs)) + for sha := range blobs { + shas = append(shas, sha) + } + correctSHA := shas[0] + corruptSHA := shas[1] + // absentSHA := shas[2] // left absent on purpose + + correctPath := filepath.Join(destDir, BlobKey(correctSHA)) + if err := os.MkdirAll(filepath.Dir(correctPath), 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(correctPath, []byte(blobs[correctSHA]), 0644); err != nil { + t.Fatal(err) + } + + corruptPath := filepath.Join(destDir, BlobKey(corruptSHA)) + if err := os.MkdirAll(filepath.Dir(corruptPath), 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(corruptPath, []byte("corrupted-content"), 0644); err != nil { + t.Fatal(err) + } + + getCountBeforePull := 0 + for _, r := range f.requestsSnapshot() { + if r.Method == "GET" && !strings.Contains(r.Range, "0-0") { + getCountBeforePull++ + } + } + + out.Reset() + errb.Reset() + code = Run([]string{"pull", "--manifests", manifestsDir, "--dest", destDir, "--target", "bunny"}, &out, &errb, getenv) + if code != 0 { + t.Fatalf("pull: expected 0, got %d (stdout=%s stderr=%s)", code, out.String(), errb.String()) + } + if !bytesContains(out.String(), "pulled=2 present=1") { + t.Fatalf("pull: expected pulled=2 present=1, got %s", out.String()) + } + + // verify the correct pre-existing file was never re-fetched with a full GET. + fullGETsForCorrect := 0 + for _, r := range f.requestsSnapshot() { + if r.Method == "GET" && strings.HasSuffix(r.Path, correctSHA) && r.Range != "bytes=0-0" { + fullGETsForCorrect++ + } + } + if fullGETsForCorrect != 0 { + t.Fatalf("expected no full GET for already-correct blob, got %d", fullGETsForCorrect) + } + + // all three blobs should now be present and correct in destDir. + for sha, content := range blobs { + got, err := os.ReadFile(filepath.Join(destDir, BlobKey(sha))) + if err != nil { + t.Fatalf("dest blob %s: %v", sha, err) + } + if string(got) != content { + t.Fatalf("dest blob %s: expected %q, got %q", sha, content, string(got)) + } + } +} diff --git a/internal/depot/local.go b/internal/depot/local.go new file mode 100644 index 0000000..eaceeae --- /dev/null +++ b/internal/depot/local.go @@ -0,0 +1,113 @@ +package depot + +import ( + "context" + "crypto/sha256" + "fmt" + "io" + "os" + "path/filepath" +) + +// LocalBackend implements Backend for local filesystem storage. +type LocalBackend struct { + Root string +} + +// Name returns the backend name. +func (b *LocalBackend) Name() string { + return "local" +} + +// Probe checks if a blob exists. +func (b *LocalBackend) Probe(ctx context.Context, sha string) (ProbeState, bool, error) { + blobPath := filepath.Join(b.Root, BlobKey(sha)) + _, err := os.Stat(blobPath) + if err == nil { + return Present, false, nil + } + if os.IsNotExist(err) { + return Absent, false, nil + } + // Real I/O error (permission, etc.) + return Absent, false, err +} + +// Put copies the file from src to the blob storage. +func (b *LocalBackend) Put(ctx context.Context, sha, src string) error { + blobPath := filepath.Join(b.Root, BlobKey(sha)) + blobDir := filepath.Dir(blobPath) + + // Create parent directories + if err := os.MkdirAll(blobDir, 0755); err != nil { + return err + } + + // Create temp file in the same directory for atomic rename + tmpFile, err := os.CreateTemp(blobDir, ".tmp-") + if err != nil { + return err + } + defer os.Remove(tmpFile.Name()) + + // Copy source to temp file + srcFile, err := os.Open(src) + if err != nil { + tmpFile.Close() + return err + } + defer srcFile.Close() + + _, err = io.Copy(tmpFile, srcFile) + tmpFile.Close() + if err != nil { + return err + } + + // Atomic rename + return os.Rename(tmpFile.Name(), blobPath) +} + +// Get fetches the blob, re-hashes it, and deletes it if the hash doesn't match. +func (b *LocalBackend) Get(ctx context.Context, sha, dest string) error { + blobPath := filepath.Join(b.Root, BlobKey(sha)) + + // Create temp file in dest directory for atomic rename + destDir := filepath.Dir(dest) + if err := os.MkdirAll(destDir, 0755); err != nil { + return err + } + + tmpFile, err := os.CreateTemp(destDir, ".tmp-") + if err != nil { + return err + } + defer os.Remove(tmpFile.Name()) + + // Copy and hash simultaneously + srcFile, err := os.Open(blobPath) + if err != nil { + tmpFile.Close() + return err + } + defer srcFile.Close() + + hasher := sha256.New() + teeReader := io.TeeReader(srcFile, hasher) + + _, err = io.Copy(tmpFile, teeReader) + tmpFile.Close() + if err != nil { + return err + } + + // Check hash + gotHash := fmt.Sprintf("%x", hasher.Sum(nil)) + if gotHash != sha { + os.Remove(tmpFile.Name()) + return fmt.Errorf("hash mismatch: expected %s, got %s", sha, gotHash) + } + + // Atomic rename + return os.Rename(tmpFile.Name(), dest) +} diff --git a/internal/depot/local_test.go b/internal/depot/local_test.go new file mode 100644 index 0000000..459a8f8 --- /dev/null +++ b/internal/depot/local_test.go @@ -0,0 +1,139 @@ +package depot + +import ( + "bytes" + "context" + "crypto/sha256" + "fmt" + "os" + "path/filepath" + "testing" +) + +func TestLocalRoundTrip(t *testing.T) { + tmpDir := t.TempDir() + backend := &LocalBackend{Root: tmpDir} + + // Generate test data + testData := []byte("hello world") + testSHA := fmt.Sprintf("%x", sha256.Sum256(testData)) + + // Create source file + srcFile := filepath.Join(tmpDir, "source.txt") + if err := os.WriteFile(srcFile, testData, 0644); err != nil { + t.Fatalf("failed to create source file: %v", err) + } + + // Put + ctx := context.Background() + if err := backend.Put(ctx, testSHA, srcFile); err != nil { + t.Fatalf("Put failed: %v", err) + } + + // Probe should be Present + state, transient, err := backend.Probe(ctx, testSHA) + if err != nil { + t.Fatalf("Probe failed: %v", err) + } + if state != Present { + t.Fatalf("expected Present, got %v", state) + } + if transient { + t.Fatalf("local backend should never be transient") + } + + // Get + destFile := filepath.Join(tmpDir, "dest.txt") + if err := backend.Get(ctx, testSHA, destFile); err != nil { + t.Fatalf("Get failed: %v", err) + } + + // Verify bytes match + gotData, err := os.ReadFile(destFile) + if err != nil { + t.Fatalf("failed to read dest file: %v", err) + } + if !bytes.Equal(gotData, testData) { + t.Fatalf("data mismatch: want %s, got %s", testData, gotData) + } +} + +func TestLocalProbeAbsent(t *testing.T) { + tmpDir := t.TempDir() + backend := &LocalBackend{Root: tmpDir} + + ctx := context.Background() + state, transient, err := backend.Probe(ctx, "0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d") + if err != nil { + t.Fatalf("Probe should not error on missing blob: %v", err) + } + if state != Absent { + t.Fatalf("expected Absent, got %v", state) + } + if transient { + t.Fatalf("local backend should never be transient") + } +} + +func TestLocalGetHashMismatch(t *testing.T) { + tmpDir := t.TempDir() + backend := &LocalBackend{Root: tmpDir} + + // Create a corrupt blob at the expected path + sha := "0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d0b1d" + blobKey := BlobKey(sha) + blobPath := filepath.Join(tmpDir, blobKey) + + if err := os.MkdirAll(filepath.Dir(blobPath), 0755); err != nil { + t.Fatalf("failed to create blob dir: %v", err) + } + if err := os.WriteFile(blobPath, []byte("wrong data"), 0644); err != nil { + t.Fatalf("failed to create blob file: %v", err) + } + + // Get should error + ctx := context.Background() + destFile := filepath.Join(tmpDir, "dest.txt") + err := backend.Get(ctx, sha, destFile) + if err == nil { + t.Fatalf("Get should error on hash mismatch") + } + + // Destination file should not exist + _, err = os.Stat(destFile) + if !os.IsNotExist(err) { + t.Fatalf("dest file should not exist after Get error: %v", err) + } +} + +func TestLocalPutIdempotent(t *testing.T) { + tmpDir := t.TempDir() + backend := &LocalBackend{Root: tmpDir} + + testData := []byte("hello world") + testSHA := fmt.Sprintf("%x", sha256.Sum256(testData)) + + srcFile := filepath.Join(tmpDir, "source.txt") + if err := os.WriteFile(srcFile, testData, 0644); err != nil { + t.Fatalf("failed to create source file: %v", err) + } + + ctx := context.Background() + + // Put twice + if err := backend.Put(ctx, testSHA, srcFile); err != nil { + t.Fatalf("first Put failed: %v", err) + } + if err := backend.Put(ctx, testSHA, srcFile); err != nil { + t.Fatalf("second Put failed: %v", err) + } + + // Verify blob exists + state, _, err := backend.Probe(ctx, testSHA) + if err != nil { + t.Fatalf("Probe failed: %v", err) + } + if state != Present { + t.Fatalf("expected Present after Put, got %v", state) + } +} diff --git a/internal/depot/manifest.go b/internal/depot/manifest.go new file mode 100644 index 0000000..6a639b6 --- /dev/null +++ b/internal/depot/manifest.go @@ -0,0 +1,76 @@ +package depot + +import ( + "fmt" + "os" + "path/filepath" + "regexp" + + "gopkg.in/yaml.v3" +) + +type manifestFile struct { + Assets []struct { + Path string `yaml:"path"` + SHA256 string `yaml:"sha256"` + Size int64 `yaml:"size"` + } `yaml:"assets"` +} + +var sha256Pattern = regexp.MustCompile(`^[0-9a-f]{64}$`) + +func ValidSHA(s string) bool { + return sha256Pattern.MatchString(s) +} + +func BlobKey(sha string) string { + return fmt.Sprintf("sha256/%s/%s/%s", sha[0:2], sha[2:4], sha) +} + +func ReferencedSHAs(dir string) (map[string]int64, error) { + entries, err := os.ReadDir(dir) + if err != nil { + return nil, err + } + + result := make(map[string]int64) + foundAny := false + + for _, entry := range entries { + if entry.IsDir() { + continue + } + if filepath.Ext(entry.Name()) != ".yml" { + continue + } + + foundAny = true + path := filepath.Join(dir, entry.Name()) + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + + var manifest manifestFile + if err := yaml.Unmarshal(data, &manifest); err != nil { + return nil, err + } + + for _, asset := range manifest.Assets { + if !ValidSHA(asset.SHA256) { + return nil, fmt.Errorf("%s: asset %q: invalid sha256 %q", entry.Name(), asset.Path, asset.SHA256) + } + + // Keep the largest size for each sha + if asset.Size > result[asset.SHA256] { + result[asset.SHA256] = asset.Size + } + } + } + + if !foundAny { + return nil, fmt.Errorf("no *.yml files found in %s", dir) + } + + return result, nil +} diff --git a/internal/depot/manifest_test.go b/internal/depot/manifest_test.go new file mode 100644 index 0000000..54f1748 --- /dev/null +++ b/internal/depot/manifest_test.go @@ -0,0 +1,129 @@ +package depot + +import ( + "os" + "path/filepath" + "testing" +) + +func TestValidSHA(t *testing.T) { + tests := []struct { + input string + want bool + }{ + {"0b1d1234567890abcdef1234567890abcdef1234567890abcdef1234567890ab", true}, + {"0000000000000000000000000000000000000000000000000000000000000000", true}, + {"ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff", true}, + {"0b1d1234567890ABCDEF1234567890abcdef1234567890abcdef1234567890ab", false}, // uppercase + {"0b1d1234567890abcde", false}, // too short + {"0b1d1234567890abcdef1234567890abcdef1234567890abcdef1234567890abff", false}, // too long + {"0b1d1234567890abcdef1234567890abcdef1234567890abcdef1234567890ag", false}, // invalid hex + {"", false}, + } + for _, tc := range tests { + got := ValidSHA(tc.input) + if got != tc.want { + t.Errorf("ValidSHA(%q): got %v, want %v", tc.input, got, tc.want) + } + } +} + +func TestBlobKey(t *testing.T) { + tests := []struct { + sha string + want string + }{ + {"0b1d1234567890abcdef1234567890abcdef1234567890abcdef1234567890ab", "sha256/0b/1d/0b1d1234567890abcdef1234567890abcdef1234567890abcdef1234567890ab"}, + {"abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789", "sha256/ab/cd/abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"}, + } + for _, tc := range tests { + got := BlobKey(tc.sha) + if got != tc.want { + t.Errorf("BlobKey(%q): got %q, want %q", tc.sha, got, tc.want) + } + } +} + +func TestReferencedSHAs(t *testing.T) { + // Test: temp dir with two yml files sharing one sha → dedup + tmpdir := t.TempDir() + + // First manifest with sha1 and sha2 + manifest1 := `assets: + - path: file1.txt + sha256: 0b1d1234567890abcdef1234567890abcdef1234567890abcdef1234567890ab + size: 100 + - path: file2.txt + sha256: abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789 + size: 200 +` + if err := os.WriteFile(filepath.Join(tmpdir, "manifest1.yml"), []byte(manifest1), 0644); err != nil { + t.Fatalf("write manifest1: %v", err) + } + + // Second manifest with sha2 and sha3 + manifest2 := `assets: + - path: file3.txt + sha256: abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789 + size: 300 + - path: file4.txt + sha256: fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210 + size: 400 +` + if err := os.WriteFile(filepath.Join(tmpdir, "manifest2.yml"), []byte(manifest2), 0644); err != nil { + t.Fatalf("write manifest2: %v", err) + } + + result, err := ReferencedSHAs(tmpdir) + if err != nil { + t.Fatalf("ReferencedSHAs: %v", err) + } + + // Should have 3 unique SHAs + if len(result) != 3 { + t.Errorf("ReferencedSHAs: got %d unique SHAs, want 3", len(result)) + } + + expectedSizes := map[string]int64{ + "0b1d1234567890abcdef1234567890abcdef1234567890abcdef1234567890ab": 100, + "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789": 300, // Largest size for duplicated sha + "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210": 400, + } + for sha, expectedSize := range expectedSizes { + size, exists := result[sha] + if !exists { + t.Errorf("ReferencedSHAs: sha %q missing", sha) + } + if size != expectedSize { + t.Errorf("ReferencedSHAs: sha %q size: got %d, want %d", sha, size, expectedSize) + } + } + + // Test: bad sha → error naming the file + badManifest := `assets: + - path: file5.txt + sha256: not_a_valid_sha + size: 500 +` + if err := os.WriteFile(filepath.Join(tmpdir, "badsha.yml"), []byte(badManifest), 0644); err != nil { + t.Fatalf("write badsha: %v", err) + } + + _, err = ReferencedSHAs(tmpdir) + if err == nil { + t.Fatalf("ReferencedSHAs with bad sha: got nil error, want error naming file and asset path") + } + want := `badsha.yml: asset "file5.txt": invalid sha256 "not_a_valid_sha"` + if err.Error() != want { + t.Errorf("ReferencedSHAs error message: got %q, want %q", err.Error(), want) + } +} + +func TestReferencedSHAsEmpty(t *testing.T) { + // Test: empty dir → error + tmpdir := t.TempDir() + _, err := ReferencedSHAs(tmpdir) + if err == nil { + t.Errorf("ReferencedSHAs on empty dir: got nil error, want error") + } +} diff --git a/internal/depot/remote.go b/internal/depot/remote.go new file mode 100644 index 0000000..cdd38dc --- /dev/null +++ b/internal/depot/remote.go @@ -0,0 +1,269 @@ +package depot + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "io" + "net" + "net/http" + "os" + "path/filepath" + "strings" + "time" +) + +// NewBackend returns the backend for name ("local"|"cdn"|"bunny"). +// localRoot is used only for "local". Fails closed: "bunny" with empty +// StorageHost or empty ReadKey returns an error (never prompts); +// unknown name returns an error. +func NewBackend(name, localRoot string, cfg Config) (Backend, error) { + switch name { + case "local": + if localRoot == "" { + return nil, errors.New("local backend requires a non-empty root") + } + return &LocalBackend{Root: localRoot}, nil + case "cdn": + return &httpBackend{ + name: "cdn", + client: newHTTPClient(cfg), + cfg: cfg, + }, nil + case "bunny": + if cfg.StorageHost == "" { + return nil, errors.New("bunny backend requires BUNNY_STORAGE_HOST") + } + if cfg.ReadKey == "" { + return nil, errors.New("bunny backend requires BUNNY_STORAGE_READ_PASSWORD or BUNNY_STORAGE_PASSWORD") + } + return &httpBackend{ + name: "bunny", + client: newHTTPClient(cfg), + cfg: cfg, + }, nil + default: + return nil, fmt.Errorf("unknown backend %q", name) + } +} + +// newHTTPClient builds an IPv4-only client per spec (HEAD probes are banned; +// IPv6 dial hazards are out of scope for this depot). +func newHTTPClient(cfg Config) *http.Client { + transport := &http.Transport{ + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + d := net.Dialer{Timeout: cfg.ConnectTimeout} + return d.DialContext(ctx, "tcp4", addr) + }, + MaxIdleConnsPerHost: cfg.ProbeJobs, + } + return &http.Client{Transport: transport, Timeout: cfg.ProbeMaxTime} +} + +// httpBackend implements Backend for both cdn (read-only) and bunny +// (read/write) over HTTP, sharing probe/get/put logic. +type httpBackend struct { + name string + client *http.Client + cfg Config +} + +func (b *httpBackend) Name() string { return b.name } + +func (b *httpBackend) storageURL(sha string) string { + host := b.cfg.StorageHost + // StorageHost is normally a bare host ("storage.bunnycdn.com"); allow a + // full scheme (used by tests against httptest.NewServer) to pass through + // unchanged. + if strings.Contains(host, "://") { + return fmt.Sprintf("%s/%s/%s", strings.TrimSuffix(host, "/"), b.cfg.StorageZone, BlobKey(sha)) + } + return fmt.Sprintf("https://%s/%s/%s", host, b.cfg.StorageZone, BlobKey(sha)) +} + +func (b *httpBackend) cdnURL(sha string) string { + return fmt.Sprintf("%s/%s", b.cfg.CDNBase, BlobKey(sha)) +} + +// probeRetrySleep is the backoff sleeper for transient probe retries; +// tests stub it (same pattern as sweep.go's confirmSleep). +var probeRetrySleep = time.Sleep + +// rangeProbe issues GET with Range: bytes=0-0 and classifies the +// response. Never HEAD (banned by spec). Transient outcomes (transport +// error, or a status that is neither 2xx nor 404/410) retry up to 2 more +// times with linear backoff (1s, 2s) — the CDN edge throttles sustained +// sweeps; the bash this ports absorbed that with curl --retry 2. +func (b *httpBackend) rangeProbe(ctx context.Context, url string, headers map[string]string) (ProbeState, bool, error) { + for attempt := 0; ; attempt++ { + state, transient, err := b.rangeProbeOnce(ctx, url, headers) + if !transient || attempt >= 2 { + return state, transient, err + } + probeRetrySleep(time.Duration(attempt+1) * time.Second) + } +} + +func (b *httpBackend) rangeProbeOnce(ctx context.Context, url string, headers map[string]string) (ProbeState, bool, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return Unconfirmed, true, err + } + req.Header.Set("Range", "bytes=0-0") + for k, v := range headers { + req.Header.Set(k, v) + } + + resp, err := b.client.Do(req) + if err != nil { + return Unconfirmed, true, err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + return Present, false, nil + case resp.StatusCode == 404 || resp.StatusCode == 410: + return Absent, false, nil + default: + return Unconfirmed, true, nil + } +} + +// Probe returns the existence state of sha at this backend. +func (b *httpBackend) Probe(ctx context.Context, sha string) (ProbeState, bool, error) { + if b.name == "cdn" { + return b.rangeProbe(ctx, b.cdnURL(sha), nil) + } + return b.rangeProbe(ctx, b.storageURL(sha), map[string]string{"AccessKey": b.cfg.ReadKey}) +} + +// Put uploads src for sha. cdn is read-only. +func (b *httpBackend) Put(ctx context.Context, sha, src string) error { + if b.name == "cdn" { + return errors.New("cdn backend is read-only") + } + if b.cfg.WriteKey == "" { + return errors.New("bunny backend requires BUNNY_STORAGE_PASSWORD to write") + } + + state, _, err := b.Probe(ctx, sha) + if err != nil { + return err + } + if state == Present { + return nil + } + + f, err := os.Open(src) + if err != nil { + return err + } + defer f.Close() + + req, err := http.NewRequestWithContext(ctx, http.MethodPut, b.storageURL(sha), f) + if err != nil { + return err + } + req.Header.Set("AccessKey", b.cfg.WriteKey) + req.Header.Set("Checksum", strings.ToUpper(sha)) + + resp, err := b.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("bunny put %s: unexpected status %d", sha, resp.StatusCode) + } + return nil +} + +// Get fetches sha into dest via temp file + rename, re-hashing and deleting +// on mismatch. Mirrors bash _depot_bunny_get: try CDN first, fall back to +// storage with ReadKey. +func (b *httpBackend) Get(ctx context.Context, sha, dest string) error { + destDir := filepath.Dir(dest) + if err := os.MkdirAll(destDir, 0755); err != nil { + return err + } + + tmpFile, err := os.CreateTemp(destDir, ".tmp-") + if err != nil { + return err + } + tmpName := tmpFile.Name() + defer os.Remove(tmpName) + + resp, err := b.fetch(ctx, sha) + if err != nil { + tmpFile.Close() + return err + } + defer resp.Body.Close() + + hasher := sha256.New() + tee := io.TeeReader(resp.Body, hasher) + _, err = io.Copy(tmpFile, tee) + tmpFile.Close() + if err != nil { + return err + } + + gotHash := fmt.Sprintf("%x", hasher.Sum(nil)) + if gotHash != sha { + os.Remove(tmpName) + return fmt.Errorf("hash mismatch: expected %s, got %s", sha, gotHash) + } + + return os.Rename(tmpName, dest) +} + +// fetch tries the CDN URL first (no auth), falling back to the storage URL +// with ReadKey on any non-2xx response or transport error. +func (b *httpBackend) fetch(ctx context.Context, sha string) (*http.Response, error) { + if b.name != "cdn" { + if resp, err := b.get(ctx, b.cdnURL(sha), nil); err == nil && resp.StatusCode >= 200 && resp.StatusCode < 300 { + return resp, nil + } else if err == nil { + _, _ = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + } + resp, err := b.get(ctx, b.storageURL(sha), map[string]string{"AccessKey": b.cfg.ReadKey}) + if err != nil { + return nil, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + _, _ = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + return nil, fmt.Errorf("bunny get %s: unexpected status %d", sha, resp.StatusCode) + } + return resp, nil + } + + resp, err := b.get(ctx, b.cdnURL(sha), nil) + if err != nil { + return nil, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + _, _ = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + return nil, fmt.Errorf("cdn get %s: unexpected status %d", sha, resp.StatusCode) + } + return resp, nil +} + +func (b *httpBackend) get(ctx context.Context, url string, headers map[string]string) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + for k, v := range headers { + req.Header.Set(k, v) + } + return b.client.Do(req) +} diff --git a/internal/depot/remote_test.go b/internal/depot/remote_test.go new file mode 100644 index 0000000..36e1b17 --- /dev/null +++ b/internal/depot/remote_test.go @@ -0,0 +1,549 @@ +package depot + +import ( + "context" + "crypto/sha256" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +func shaOf(s string) string { + return fmt.Sprintf("%x", sha256.Sum256([]byte(s))) +} + +type recordedReq struct { + Method string + Path string + Range string + Headers http.Header +} + +// fakeBunny is a minimal recording fake for Bunny storage/CDN endpoints. +type fakeBunny struct { + mu sync.Mutex + requests []recordedReq + headCount int + blobs map[string][]byte + statusFor map[string]int // sha -> status code override for probe/get +} + +func newFakeBunny() *fakeBunny { + return &fakeBunny{blobs: map[string][]byte{}, statusFor: map[string]int{}} +} + +func (f *fakeBunny) handler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + f.requests = append(f.requests, recordedReq{ + Method: r.Method, + Path: r.URL.Path, + Range: r.Header.Get("Range"), + Headers: r.Header.Clone(), + }) + if r.Method == http.MethodHead { + f.headCount++ + f.mu.Unlock() + w.WriteHeader(500) + return + } + f.mu.Unlock() + + sha := strings.TrimPrefix(r.URL.Path, "/") + // last path element is the sha + parts := strings.Split(sha, "/") + sha = parts[len(parts)-1] + + if r.Method == http.MethodPut { + buf, _ := io.ReadAll(r.Body) + f.mu.Lock() + f.blobs[sha] = buf + f.mu.Unlock() + w.WriteHeader(201) + return + } + + // GET (range probe / actual get) + f.mu.Lock() + status, hasStatus := f.statusFor[sha] + data, ok := f.blobs[sha] + f.mu.Unlock() + + if hasStatus { + w.WriteHeader(status) + return + } + if !ok { + w.WriteHeader(404) + return + } + w.WriteHeader(206) + _, _ = w.Write(data) + } +} + +func (f *fakeBunny) requestsSnapshot() []recordedReq { + f.mu.Lock() + defer f.mu.Unlock() + out := make([]recordedReq, len(f.requests)) + copy(out, f.requests) + return out +} + +func testCfg(storageHost, readKey, writeKey string) Config { + return Config{ + CDNBase: "http://127.0.0.1:1", // unreachable by default; overridden per test + StorageHost: storageHost, + StorageZone: "sow-assets-depot", + ReadKey: readKey, + WriteKey: writeKey, + ProbeJobs: 4, + Jobs: 4, + ConnectTimeout: 2 * time.Second, + ProbeMaxTime: 2 * time.Second, + ConfirmRetries: 3, + ConfirmMax: 200, + } +} + +// hostPort returns a StorageHost value for tests: a full "http://host:port" +// base URL, which storageURL() passes through unchanged (bypassing the +// production https:// default so httptest.NewServer works without TLS). +func hostPort(srv *httptest.Server) string { + return srv.URL +} + +// stubProbeSleep replaces probeRetrySleep for the test, recording durations. +func stubProbeSleep(t *testing.T) *[]time.Duration { + t.Helper() + var slept []time.Duration + old := probeRetrySleep + probeRetrySleep = func(d time.Duration) { slept = append(slept, d) } + t.Cleanup(func() { probeRetrySleep = old }) + return &slept +} + +func TestProbeRetriesTransientThenPresent(t *testing.T) { + slept := stubProbeSleep(t) + sha := shaOf("throttled-blob") + + var calls int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + if calls == 1 { + w.WriteHeader(428) + return + } + w.WriteHeader(206) + })) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + state, transient, err := b.Probe(context.Background(), sha) + if err != nil { + t.Fatalf("Probe: %v", err) + } + if state != Present || transient { + t.Fatalf("expected Present/non-transient after retry, got %v/%v", state, transient) + } + if calls != 2 { + t.Fatalf("expected 2 requests (428 then 206), got %d", calls) + } + if len(*slept) != 1 || (*slept)[0] != time.Second { + t.Fatalf("expected one 1s backoff, got %v", *slept) + } +} + +func TestProbe404NoRetry(t *testing.T) { + slept := stubProbeSleep(t) + sha := shaOf("gone-blob") + + f := newFakeBunny() + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + state, transient, err := b.Probe(context.Background(), sha) + if err != nil { + t.Fatalf("Probe: %v", err) + } + if state != Absent || transient { + t.Fatalf("expected Absent/non-transient, got %v/%v", state, transient) + } + if got := len(f.requestsSnapshot()); got != 1 { + t.Fatalf("expected exactly 1 request for 404, got %d", got) + } + if len(*slept) != 0 { + t.Fatalf("expected zero backoffs for 404, got %v", *slept) + } +} + +func TestProbeUsesRangeGetNotHead(t *testing.T) { + sha := shaOf("present-blob") + f := newFakeBunny() + f.blobs[sha] = []byte("hello") + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + state, transient, err := b.Probe(context.Background(), sha) + if err != nil { + t.Fatalf("Probe: %v", err) + } + if state != Present || transient { + t.Fatalf("expected Present/non-transient, got %v/%v", state, transient) + } + + reqs := f.requestsSnapshot() + if len(reqs) != 1 { + t.Fatalf("expected exactly 1 request, got %d", len(reqs)) + } + if reqs[0].Method != "GET" { + t.Fatalf("expected GET, got %s", reqs[0].Method) + } + if reqs[0].Range != "bytes=0-0" { + t.Fatalf("expected Range bytes=0-0, got %q", reqs[0].Range) + } + if f.headCount != 0 { + t.Fatalf("expected zero HEAD requests, got %d", f.headCount) + } +} + +func TestProbeStates(t *testing.T) { + stubProbeSleep(t) + cases := []struct { + status int + wantState ProbeState + wantTransient bool + }{ + {200, Present, false}, + {206, Present, false}, + {404, Absent, false}, + {428, Unconfirmed, true}, + {503, Unconfirmed, true}, + } + + for _, c := range cases { + t.Run(fmt.Sprintf("status_%d", c.status), func(t *testing.T) { + sha := shaOf(fmt.Sprintf("blob-%d", c.status)) + f := newFakeBunny() + f.statusFor[sha] = c.status + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + state, transient, err := b.Probe(context.Background(), sha) + if err != nil && c.status < 500 { + t.Fatalf("unexpected error: %v", err) + } + if state != c.wantState { + t.Fatalf("status %d: expected state %v, got %v", c.status, c.wantState, state) + } + if transient != c.wantTransient { + t.Fatalf("status %d: expected transient=%v, got %v", c.status, c.wantTransient, transient) + } + if f.headCount != 0 { + t.Fatalf("expected zero HEAD requests, got %d", f.headCount) + } + }) + } +} + +func TestBunnyPutSequence(t *testing.T) { + t.Run("absent sha", func(t *testing.T) { + sha := shaOf("new-blob") + f := newFakeBunny() + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + srcPath := filepath.Join(dir, "src") + if err := os.WriteFile(srcPath, []byte("new-blob-content"), 0644); err != nil { + t.Fatal(err) + } + + if err := b.Put(context.Background(), sha, srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + reqs := f.requestsSnapshot() + if len(reqs) != 2 { + t.Fatalf("expected 2 requests (probe, put), got %d: %+v", len(reqs), reqs) + } + if reqs[0].Method != "GET" || reqs[0].Range != "bytes=0-0" { + t.Fatalf("expected first request to be range probe GET, got %+v", reqs[0]) + } + if reqs[1].Method != "PUT" { + t.Fatalf("expected second request to be PUT, got %+v", reqs[1]) + } + if got := reqs[1].Headers.Get("Checksum"); got != strings.ToUpper(sha) { + t.Fatalf("expected Checksum %s, got %s", strings.ToUpper(sha), got) + } + if got := reqs[1].Headers.Get("AccessKey"); got != "writekey" { + t.Fatalf("expected AccessKey writekey, got %s", got) + } + if f.headCount != 0 { + t.Fatalf("expected zero HEAD requests, got %d", f.headCount) + } + }) + + t.Run("present sha", func(t *testing.T) { + sha := shaOf("existing-blob") + f := newFakeBunny() + f.blobs[sha] = []byte("existing-blob-content") + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + srcPath := filepath.Join(dir, "src") + if err := os.WriteFile(srcPath, []byte("existing-blob-content"), 0644); err != nil { + t.Fatal(err) + } + + if err := b.Put(context.Background(), sha, srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + reqs := f.requestsSnapshot() + if len(reqs) != 1 { + t.Fatalf("expected probe-only (1 request) since present, got %d: %+v", len(reqs), reqs) + } + if reqs[0].Method != "GET" { + t.Fatalf("expected GET probe, got %+v", reqs[0]) + } + }) +} + +func TestReadWriteKeySplit(t *testing.T) { + sha := shaOf("split-blob") + f := newFakeBunny() + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkeyXXX", "writekeyYYY") + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + srcPath := filepath.Join(dir, "src") + if err := os.WriteFile(srcPath, []byte("split-blob-content"), 0644); err != nil { + t.Fatal(err) + } + + if err := b.Put(context.Background(), sha, srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + reqs := f.requestsSnapshot() + if len(reqs) != 2 { + t.Fatalf("expected 2 requests, got %d", len(reqs)) + } + if got := reqs[0].Headers.Get("AccessKey"); got != "readkeyXXX" { + t.Fatalf("probe expected AccessKey readkeyXXX, got %s", got) + } + if got := reqs[1].Headers.Get("AccessKey"); got != "writekeyYYY" { + t.Fatalf("put expected AccessKey writekeyYYY, got %s", got) + } + if reqs[0].Headers.Get("AccessKey") == reqs[1].Headers.Get("AccessKey") { + t.Fatalf("expected distinct read/write keys") + } +} + +func TestBunnyNoReadKeyFailsClosed(t *testing.T) { + // Grep-based tripwire: no non-test depot source may reference stdin/tty. + dir := "." + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("ReadDir: %v", err) + } + for _, e := range entries { + name := e.Name() + if e.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") { + continue + } + data, err := os.ReadFile(filepath.Join(dir, name)) + if err != nil { + t.Fatalf("ReadFile %s: %v", name, err) + } + s := string(data) + if strings.Contains(s, "os.Stdin") || strings.Contains(s, "bufio.NewReader(os.Stdin)") || strings.Contains(s, "/dev/tty") { + t.Fatalf("%s references stdin/tty; credential prompts are banned", name) + } + } + + // Confirm undrained pipe: swap in an os.Pipe as stdin-equivalent to prove + // no read occurs. + r, w, err := os.Pipe() + if err != nil { + t.Fatalf("Pipe: %v", err) + } + oldStdin := os.Stdin + os.Stdin = r + defer func() { os.Stdin = oldStdin; r.Close() }() + + cfg := testCfg("storage.example.com", "", "writekey") + _, err = NewBackend("bunny", "", cfg) + if err == nil { + t.Fatal("expected error for empty ReadKey") + } + if !strings.Contains(err.Error(), "ReadKey") && !strings.Contains(err.Error(), "BUNNY_STORAGE_READ_PASSWORD") && !strings.Contains(err.Error(), "BUNNY_STORAGE_PASSWORD") { + t.Fatalf("expected error to mention missing env var, got: %v", err) + } + + w.Close() + // If code read from stdin it would have blocked already (pipe with no + // writer-side data yet); assert the pipe is still open/undrained by + // writing now and reading it back ourselves. + if _, err := w.Write([]byte("x")); err == nil { + t.Fatal("expected write to closed pipe writer to fail") + } +} + +func TestBunnyNoStorageHostFailsClosed(t *testing.T) { + cfg := testCfg("", "readkey", "writekey") + _, err := NewBackend("bunny", "", cfg) + if err == nil { + t.Fatal("expected error for empty StorageHost") + } + if !strings.Contains(err.Error(), "StorageHost") && !strings.Contains(err.Error(), "BUNNY_STORAGE_HOST") { + t.Fatalf("expected error to mention missing StorageHost/BUNNY_STORAGE_HOST, got: %v", err) + } +} + +func TestGetHashMismatchDeletes(t *testing.T) { + sha := shaOf("get-mismatch-blob") + f := newFakeBunny() + f.blobs[sha] = []byte("wrong content") + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + cfg := testCfg(hostPort(srv), "readkey", "writekey") + // force CDN unreachable so Get falls back to storage + cfg.CDNBase = "http://127.0.0.1:1" + b, err := NewBackend("bunny", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + dest := filepath.Join(dir, "dest") + + err = b.Get(context.Background(), sha, dest) + if err == nil { + t.Fatal("expected hash mismatch error") + } + if _, statErr := os.Stat(dest); !os.IsNotExist(statErr) { + t.Fatalf("expected dest to be removed on mismatch, stat err: %v", statErr) + } +} + +func TestCDNPutReadOnly(t *testing.T) { + cfg := testCfg("storage.example.com", "readkey", "writekey") + b, err := NewBackend("cdn", "", cfg) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + + dir := t.TempDir() + srcPath := filepath.Join(dir, "src") + if err := os.WriteFile(srcPath, []byte("data"), 0644); err != nil { + t.Fatal(err) + } + + err = b.Put(context.Background(), shaOf("anything"), srcPath) + if err == nil { + t.Fatal("expected error for cdn Put (read-only)") + } + if !strings.Contains(err.Error(), "read-only") { + t.Fatalf("expected read-only error, got: %v", err) + } +} + +// TestNoHeadInSources is the anti-HEAD regression tripwire from field note 3: +// grep-assert no http.MethodHead/.Head( usage in non-test depot sources. +func TestNoHeadInSources(t *testing.T) { + dir := "." + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("ReadDir: %v", err) + } + for _, e := range entries { + name := e.Name() + if e.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") { + continue + } + data, err := os.ReadFile(filepath.Join(dir, name)) + if err != nil { + t.Fatalf("ReadFile %s: %v", name, err) + } + s := string(data) + if strings.Contains(s, "http.MethodHead") || strings.Contains(s, ".Head(") { + t.Fatalf("%s references HEAD probe (banned)", name) + } + } +} + +// TestLocalBackendViaFactory sanity-checks NewBackend("local", ...). +func TestLocalBackendViaFactory(t *testing.T) { + root := t.TempDir() + b, err := NewBackend("local", root, Config{}) + if err != nil { + t.Fatalf("NewBackend: %v", err) + } + lb, ok := b.(*LocalBackend) + if !ok { + t.Fatalf("expected *LocalBackend, got %T", b) + } + if lb.Root != root { + t.Fatalf("expected Root=%s, got %s", root, lb.Root) + } + + if _, err := NewBackend("local", "", Config{}); err == nil { + t.Fatal("expected error for empty local root") + } + + if _, err := NewBackend("nonsense", "", Config{}); err == nil { + t.Fatal("expected error for unknown backend name") + } +} diff --git a/internal/depot/run.go b/internal/depot/run.go new file mode 100644 index 0000000..1c49876 --- /dev/null +++ b/internal/depot/run.go @@ -0,0 +1,479 @@ +package depot + +import ( + "context" + "crypto/sha256" + "errors" + "flag" + "fmt" + "io" + "math/rand" + "os" + "path/filepath" + "sort" + "sync" + "time" +) + +const ( + exitOK = 0 + exitDrift = 1 + exitUnconfirmed = 2 + exitUsage = 64 + exitInternal = 70 +) + +// Run executes a depot subcommand. args[0] is the subcommand +// (status|push|verify|get|pull); returns the process exit code. +func Run(args []string, stdout, stderr io.Writer, getenv func(string) string) int { + if len(args) == 0 { + printRunUsage(stderr) + return exitUsage + } + rest := args[1:] + switch args[0] { + case "status": + return runStatus(rest, stdout, stderr, getenv) + case "push": + return runPush(rest, stdout, stderr, getenv) + case "verify": + return runVerify(rest, stdout, stderr, getenv) + case "get": + return runGet(rest, stdout, stderr, getenv) + case "pull": + return runPull(rest, stdout, stderr, getenv) + default: + fmt.Fprintf(stderr, "depot: unknown subcommand %q\n\n", args[0]) + printRunUsage(stderr) + return exitUsage + } +} + +func printRunUsage(w io.Writer) { + fmt.Fprint(w, `usage: + depot status [--manifests DIR] [--source LOCAL_DEPOT] --target bunny|cdn|local + depot push [--manifests DIR] --source LOCAL_DEPOT --target bunny + depot verify [--manifests DIR] --target bunny|cdn [--sample N] + depot get --target cdn|bunny|local + depot pull [--manifests DIR] --dest DIR --target cdn|bunny + +--target local uses the DEPOT_DIR environment variable as the local root. +`) +} + +// errUsage marks errors that are the caller's fault (usage, exit 64) rather +// than internal/backend failures (exit 70). +var errUsage = errors.New("usage error") + +// resolveBackend builds the named backend, resolving "local" against DEPOT_DIR. +func resolveBackend(target string, getenv func(string) string, cfg Config) (Backend, error) { + root := "" + if target == "local" { + root = getenv("DEPOT_DIR") + if root == "" { + return nil, fmt.Errorf("--target local requires DEPOT_DIR to be set: %w", errUsage) + } + } + return NewBackend(target, root, cfg) +} + +// backendErrExit maps a resolveBackend error onto the exit contract. +func backendErrExit(err error) int { + if errors.Is(err, errUsage) { + return exitUsage + } + return exitInternal +} + +func printSweepStatus(stdout io.Writer, referenced int, res SweepResult) { + fmt.Fprintf(stdout, "referenced=%d present=%d missing=%d unconfirmed=%d\n", + referenced, len(res.Present), len(res.Missing), len(res.Unconfirmed)) + for _, sha := range res.Missing { + fmt.Fprintf(stdout, "missing %s\n", sha) + } + for _, sha := range res.Unconfirmed { + fmt.Fprintf(stdout, "unconfirmed %s\n", sha) + } +} + +func sweepExitCode(res SweepResult) int { + if len(res.Missing) > 0 { + return exitDrift + } + if len(res.Unconfirmed) > 0 { + return exitUnconfirmed + } + return exitOK +} + +func shaKeys(m map[string]int64) []string { + out := make([]string, 0, len(m)) + for sha := range m { + out = append(out, sha) + } + sort.Strings(out) + return out +} + +func runStatus(args []string, stdout, stderr io.Writer, getenv func(string) string) int { + fs := flag.NewFlagSet("status", flag.ContinueOnError) + fs.SetOutput(stderr) + manifests := fs.String("manifests", "assets", "directory of *.yml manifests") + _ = fs.String("source", "", "local depot root (unused for status target=local; see DEPOT_DIR)") + target := fs.String("target", "", "bunny|cdn|local") + if err := fs.Parse(args); err != nil { + return exitUsage + } + if *target == "" { + fmt.Fprintln(stderr, "depot status: --target is required") + return exitUsage + } + + cfg := LoadConfig(getenv) + backend, err := resolveBackend(*target, getenv, cfg) + if err != nil { + fmt.Fprintln(stderr, "depot status:", err) + return backendErrExit(err) + } + + shaSizes, err := ReferencedSHAs(*manifests) + if err != nil { + fmt.Fprintln(stderr, "depot status:", err) + return exitInternal + } + shas := shaKeys(shaSizes) + + res, err := Sweep(context.Background(), backend, shas, cfg, io.Discard) + if err != nil { + fmt.Fprintln(stderr, "depot status:", err) + return exitInternal + } + + printSweepStatus(stdout, len(shas), res) + return sweepExitCode(res) +} + +func runPush(args []string, stdout, stderr io.Writer, getenv func(string) string) int { + fs := flag.NewFlagSet("push", flag.ContinueOnError) + fs.SetOutput(stderr) + manifests := fs.String("manifests", "assets", "directory of *.yml manifests") + source := fs.String("source", "", "local depot root to read blobs from") + target := fs.String("target", "", "must be bunny") + if err := fs.Parse(args); err != nil { + return exitUsage + } + if *target != "bunny" { + fmt.Fprintln(stderr, "depot push: --target must be bunny") + return exitUsage + } + if *source == "" { + fmt.Fprintln(stderr, "depot push: --source is required") + return exitUsage + } + + cfg := LoadConfig(getenv) + backend, err := NewBackend("bunny", "", cfg) + if err != nil { + fmt.Fprintln(stderr, "depot push:", err) + return exitInternal + } + + shaSizes, err := ReferencedSHAs(*manifests) + if err != nil { + fmt.Fprintln(stderr, "depot push:", err) + return exitInternal + } + shas := shaKeys(shaSizes) + + res, err := Sweep(context.Background(), backend, shas, cfg, io.Discard) + if err != nil { + fmt.Fprintln(stderr, "depot push:", err) + return exitInternal + } + + toUpload := append(append([]string(nil), res.Missing...), res.Unconfirmed...) + sort.Strings(toUpload) + + var ( + mu sync.Mutex + uploaded int + failed int + transportErr error + ) + + work := func(sha string) { + srcPath := filepath.Join(*source, BlobKey(sha)) + if _, statErr := os.Stat(srcPath); statErr != nil { + mu.Lock() + failed++ + mu.Unlock() + fmt.Fprintf(stderr, "push: missing source blob %s\n", sha) + return + } + if err := backend.Put(context.Background(), sha, srcPath); err != nil { + mu.Lock() + failed++ + if transportErr == nil { + transportErr = err + } + mu.Unlock() + fmt.Fprintf(stderr, "push: upload %s: %v\n", sha, err) + return + } + mu.Lock() + uploaded++ + mu.Unlock() + } + + runWorkers(cfg.Jobs, toUpload, work) + + fmt.Fprintf(stdout, "uploaded=%d failed=%d\n", uploaded, failed) + + if transportErr != nil { + return exitInternal + } + if failed > 0 { + return exitDrift + } + return exitOK +} + +func runVerify(args []string, stdout, stderr io.Writer, getenv func(string) string) int { + fs := flag.NewFlagSet("verify", flag.ContinueOnError) + fs.SetOutput(stderr) + manifests := fs.String("manifests", "assets", "directory of *.yml manifests") + target := fs.String("target", "", "bunny|cdn") + sample := fs.Int("sample", 3, "number of present blobs to spot-check by download (0=skip)") + if err := fs.Parse(args); err != nil { + return exitUsage + } + if *target != "bunny" && *target != "cdn" { + fmt.Fprintln(stderr, "depot verify: --target must be bunny or cdn") + return exitUsage + } + + cfg := LoadConfig(getenv) + backend, err := NewBackend(*target, "", cfg) + if err != nil { + fmt.Fprintln(stderr, "depot verify:", err) + return exitInternal + } + + shaSizes, err := ReferencedSHAs(*manifests) + if err != nil { + fmt.Fprintln(stderr, "depot verify:", err) + return exitInternal + } + shas := shaKeys(shaSizes) + + res, err := Sweep(context.Background(), backend, shas, cfg, io.Discard) + if err != nil { + fmt.Fprintln(stderr, "depot verify:", err) + return exitInternal + } + + printSweepStatus(stdout, len(shas), res) + if code := sweepExitCode(res); code != exitOK { + return code + } + + if *sample <= 0 || len(res.Present) == 0 { + return exitOK + } + + n := *sample + if n > len(res.Present) { + n = len(res.Present) + } + rng := rand.New(rand.NewSource(time.Now().UnixNano())) + picks := rng.Perm(len(res.Present))[:n] + + tmpDir, err := os.MkdirTemp("", "depot-verify-") + if err != nil { + fmt.Fprintln(stderr, "depot verify:", err) + return exitInternal + } + defer os.RemoveAll(tmpDir) + + for _, idx := range picks { + sha := res.Present[idx] + dest := filepath.Join(tmpDir, sha) + if err := backend.Get(context.Background(), sha, dest); err != nil { + fmt.Fprintf(stderr, "depot verify: sample %s: %v\n", sha, err) + return exitDrift + } + } + return exitOK +} + +func runGet(args []string, stdout, stderr io.Writer, getenv func(string) string) int { + fs := flag.NewFlagSet("get", flag.ContinueOnError) + fs.SetOutput(stderr) + target := fs.String("target", "", "cdn|bunny|local") + if err := fs.Parse(args); err != nil { + return exitUsage + } + positional := fs.Args() + if len(positional) != 2 { + fmt.Fprintln(stderr, "depot get: usage: depot get --target cdn|bunny|local") + return exitUsage + } + sha, dest := positional[0], positional[1] + if !ValidSHA(sha) { + fmt.Fprintf(stderr, "depot get: invalid sha %q\n", sha) + return exitUsage + } + if *target == "" { + fmt.Fprintln(stderr, "depot get: --target is required") + return exitUsage + } + + cfg := LoadConfig(getenv) + backend, err := resolveBackend(*target, getenv, cfg) + if err != nil { + fmt.Fprintln(stderr, "depot get:", err) + return backendErrExit(err) + } + + if err := backend.Get(context.Background(), sha, dest); err != nil { + fmt.Fprintln(stderr, "depot get:", err) + return exitInternal + } + return exitOK +} + +func runPull(args []string, stdout, stderr io.Writer, getenv func(string) string) int { + fs := flag.NewFlagSet("pull", flag.ContinueOnError) + fs.SetOutput(stderr) + manifests := fs.String("manifests", "assets", "directory of *.yml manifests") + dest := fs.String("dest", "", "local directory to pull blobs into") + target := fs.String("target", "", "cdn|bunny") + if err := fs.Parse(args); err != nil { + return exitUsage + } + if *target != "bunny" && *target != "cdn" { + fmt.Fprintln(stderr, "depot pull: --target must be bunny or cdn") + return exitUsage + } + if *dest == "" { + fmt.Fprintln(stderr, "depot pull: --dest is required") + return exitUsage + } + + cfg := LoadConfig(getenv) + backend, err := NewBackend(*target, "", cfg) + if err != nil { + fmt.Fprintln(stderr, "depot pull:", err) + return exitInternal + } + + shaSizes, err := ReferencedSHAs(*manifests) + if err != nil { + fmt.Fprintln(stderr, "depot pull:", err) + return exitInternal + } + shas := shaKeys(shaSizes) + + // Split into "already present and correct locally" (skip) vs "needs a + // probe/download decision". + var present int + var toCheck []string + for _, sha := range shas { + destPath := filepath.Join(*dest, BlobKey(sha)) + if localFileMatchesSHA(destPath, sha) { + present++ + continue + } + toCheck = append(toCheck, sha) + } + + // Sweep the source to distinguish "not there" (exit 1, listed) from + // "there, download it". + res, err := Sweep(context.Background(), backend, toCheck, cfg, io.Discard) + if err != nil { + fmt.Fprintln(stderr, "depot pull:", err) + return exitInternal + } + + var ( + mu sync.Mutex + pulled int + anyFail bool + ) + work := func(sha string) { + destPath := filepath.Join(*dest, BlobKey(sha)) + if err := backend.Get(context.Background(), sha, destPath); err != nil { + mu.Lock() + anyFail = true + mu.Unlock() + fmt.Fprintf(stderr, "pull: %s: %v\n", sha, err) + return + } + mu.Lock() + pulled++ + mu.Unlock() + } + runWorkers(cfg.Jobs, res.Present, work) + + // ponytail: unconfirmed source state (probe budget exhausted / flaky + // responses) is treated as a download failure rather than a third exit + // path; if that proves too coarse in practice, give pull its own + // unconfirmed accounting like status. + if len(res.Unconfirmed) > 0 { + anyFail = true + for _, sha := range res.Unconfirmed { + fmt.Fprintf(stderr, "pull: %s: unconfirmed at source\n", sha) + } + } + + for _, sha := range res.Missing { + fmt.Fprintf(stdout, "missing %s\n", sha) + } + + fmt.Fprintf(stdout, "pulled=%d present=%d\n", pulled, present) + + if anyFail { + return exitInternal + } + if len(res.Missing) > 0 { + return exitDrift + } + return exitOK +} + +// localFileMatchesSHA reports whether path exists and hashes to sha. +func localFileMatchesSHA(path, sha string) bool { + f, err := os.Open(path) + if err != nil { + return false + } + defer f.Close() + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return false + } + return fmt.Sprintf("%x", h.Sum(nil)) == sha +} + +// runWorkers runs work(sha) for each sha in items using up to jobs goroutines. +func runWorkers(jobs int, items []string, work func(sha string)) { + if jobs < 1 { + jobs = 1 + } + ch := make(chan string) + var wg sync.WaitGroup + for i := 0; i < jobs; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for sha := range ch { + work(sha) + } + }() + } + for _, sha := range items { + ch <- sha + } + close(ch) + wg.Wait() +} diff --git a/internal/depot/run_test.go b/internal/depot/run_test.go new file mode 100644 index 0000000..a2bcbda --- /dev/null +++ b/internal/depot/run_test.go @@ -0,0 +1,181 @@ +package depot + +import ( + "bytes" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" +) + +func testGetenv(vals map[string]string) func(string) string { + return func(k string) string { return vals[k] } +} + +func TestRunUsage(t *testing.T) { + t.Run("no args", func(t *testing.T) { + var out, errb bytes.Buffer + code := Run(nil, &out, &errb, testGetenv(nil)) + if code != 64 { + t.Fatalf("expected 64, got %d (stderr=%s)", code, errb.String()) + } + }) + + t.Run("unknown subcommand", func(t *testing.T) { + var out, errb bytes.Buffer + code := Run([]string{"bogus"}, &out, &errb, testGetenv(nil)) + if code != 64 { + t.Fatalf("expected 64, got %d (stderr=%s)", code, errb.String()) + } + }) + + t.Run("status --target local without DEPOT_DIR", func(t *testing.T) { + dir := t.TempDir() + writeManifest(t, dir, shaOf("blob-a"), 5) + var out, errb bytes.Buffer + code := Run([]string{"status", "--manifests", dir, "--target", "local"}, &out, &errb, testGetenv(nil)) + if code != 64 { + t.Fatalf("expected 64, got %d (stderr=%s)", code, errb.String()) + } + if !bytesContains(errb.String(), "DEPOT_DIR") { + t.Fatalf("expected stderr to mention DEPOT_DIR, got %s", errb.String()) + } + }) + + t.Run("push --target cdn rejected", func(t *testing.T) { + dir := t.TempDir() + var out, errb bytes.Buffer + code := Run([]string{"push", "--manifests", dir, "--source", dir, "--target", "cdn"}, &out, &errb, testGetenv(nil)) + if code != 64 { + t.Fatalf("expected 64, got %d (stderr=%s)", code, errb.String()) + } + }) +} + +func TestGetInvalidSHA(t *testing.T) { + var out, errb bytes.Buffer + dir := t.TempDir() + code := Run([]string{"get", "not-a-sha", dir + "/dest", "--target", "local"}, &out, &errb, testGetenv(map[string]string{"DEPOT_DIR": dir})) + if code != 64 { + t.Fatalf("expected 64, got %d (stderr=%s)", code, errb.String()) + } +} + +func TestNoKeyStatusBunnyFailsClosedNoStdin(t *testing.T) { + dir := t.TempDir() + writeManifest(t, dir, shaOf("blob-a"), 5) + + var out, errb bytes.Buffer + code := Run([]string{"status", "--manifests", dir, "--target", "bunny"}, &out, &errb, testGetenv(nil)) + if code != 70 { + t.Fatalf("expected 70, got %d (stdout=%s stderr=%s)", code, out.String(), errb.String()) + } + if !bytesContains(errb.String(), "BUNNY_STORAGE_HOST") { + t.Fatalf("expected stderr to mention BUNNY_STORAGE_HOST, got %s", errb.String()) + } +} + +func TestStatusExitCodes(t *testing.T) { + t.Run("drift", func(t *testing.T) { + f := newFakeBunny() + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + sha := shaOf("missing-blob") + dir := t.TempDir() + writeManifest(t, dir, sha, 5) + + var out, errb bytes.Buffer + code := Run([]string{"status", "--manifests", dir, "--target", "bunny"}, &out, &errb, + testGetenv(map[string]string{ + "BUNNY_STORAGE_HOST": hostPort(srv), + "BUNNY_STORAGE_READ_PASSWORD": "readkey", + })) + if code != 1 { + t.Fatalf("expected 1, got %d (stdout=%s stderr=%s)", code, out.String(), errb.String()) + } + if !bytesContains(out.String(), "missing="+"1") { + t.Fatalf("expected missing=1 in output, got %s", out.String()) + } + if !bytesContains(out.String(), "missing "+sha) { + t.Fatalf("expected missing line for sha, got %s", out.String()) + } + }) + + t.Run("unconfirmed-only", func(t *testing.T) { + stubProbeSleep(t) + f := newFakeBunny() + sha := shaOf("flaky-blob") + f.statusFor[sha] = 428 + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + dir := t.TempDir() + writeManifest(t, dir, sha, 5) + + var out, errb bytes.Buffer + code := Run([]string{"status", "--manifests", dir, "--target", "bunny"}, &out, &errb, + testGetenv(map[string]string{ + "BUNNY_STORAGE_HOST": hostPort(srv), + "BUNNY_STORAGE_READ_PASSWORD": "readkey", + "DEPOT_CONFIRM_RETRIES": "1", + })) + if code != 2 { + t.Fatalf("expected 2, got %d (stdout=%s stderr=%s)", code, out.String(), errb.String()) + } + if !bytesContains(out.String(), "unconfirmed="+"1") { + t.Fatalf("expected unconfirmed=1 in output, got %s", out.String()) + } + }) +} + +func TestPushTransportErrorCountsFailed(t *testing.T) { + // Probes 404 (blob absent), PUTs 500 (transport-level upload failure). + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPut { + w.WriteHeader(500) + return + } + w.WriteHeader(404) + })) + defer srv.Close() + + sha := shaOf("push-fail-blob") + manifestsDir := t.TempDir() + writeManifest(t, manifestsDir, sha, 14) + sourceDir := t.TempDir() + blobPath := filepath.Join(sourceDir, BlobKey(sha)) + if err := os.MkdirAll(filepath.Dir(blobPath), 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(blobPath, []byte("push-fail-blob"), 0644); err != nil { + t.Fatal(err) + } + + var out, errb bytes.Buffer + code := Run([]string{"push", "--manifests", manifestsDir, "--source", sourceDir, "--target", "bunny"}, &out, &errb, + testGetenv(map[string]string{ + "BUNNY_STORAGE_HOST": hostPort(srv), + "BUNNY_STORAGE_PASSWORD": "writekey", + })) + if code != 70 { + t.Fatalf("expected 70, got %d (stdout=%s stderr=%s)", code, out.String(), errb.String()) + } + if !bytesContains(out.String(), "uploaded=0 failed=1") { + t.Fatalf("expected uploaded=0 failed=1, got %s", out.String()) + } +} + +func bytesContains(s, substr string) bool { + return bytes.Contains([]byte(s), []byte(substr)) +} + +func writeManifest(t *testing.T, dir, sha string, size int64) { + t.Helper() + content := fmt.Sprintf("assets:\n - path: foo\n sha256: %s\n size: %d\n", sha, size) + if err := os.WriteFile(filepath.Join(dir, "manifest.yml"), []byte(content), 0644); err != nil { + t.Fatal(err) + } +} diff --git a/internal/depot/sweep.go b/internal/depot/sweep.go new file mode 100644 index 0000000..81a3d20 --- /dev/null +++ b/internal/depot/sweep.go @@ -0,0 +1,156 @@ +package depot + +import ( + "context" + "fmt" + "io" + "sort" + "sync" + "time" +) + +// confirmSleep is the backoff sleeper for the serial confirm phase; +// overridden in tests. +var confirmSleep = time.Sleep + +// SweepResult partitions the swept shas by confirmed state. Unconfirmed is +// distinct from Missing and must never be collapsed into it: it means the +// probe budget was exhausted, not that the blob is known absent. +type SweepResult struct { + Present []string + Missing []string + Unconfirmed []string +} + +type probeOutcome struct { + sha string + state ProbeState + transient bool +} + +// Sweep probes every sha against b: one parallel pass (cfg.ProbeJobs workers) +// followed by a serial confirm phase over every non-Present result (unless +// the non-Present count exceeds cfg.ConfirmMax, in which case all of them +// are reported Unconfirmed with zero re-probes). Progress is reported to +// progress roughly every 1000 blobs. +func Sweep(ctx context.Context, b Backend, shas []string, cfg Config, progress io.Writer) (SweepResult, error) { + sorted := append([]string(nil), shas...) + sort.Strings(sorted) + + outcomes := make([]probeOutcome, len(sorted)) + jobs := cfg.ProbeJobs + if jobs < 1 { + jobs = 1 + } + + var wg sync.WaitGroup + work := make(chan int) + var probed int64 + var mu sync.Mutex + var firstErr error + + worker := func() { + defer wg.Done() + for i := range work { + state, transient, err := b.Probe(ctx, sorted[i]) + mu.Lock() + if err != nil && !transient { + if firstErr == nil { + firstErr = fmt.Errorf("probe %s: %w", sorted[i], err) + } + } else { + outcomes[i] = probeOutcome{sha: sorted[i], state: state, transient: transient} + } + probed++ + n := probed + mu.Unlock() + if n%1000 == 0 { + fmt.Fprintf(progress, "probed %d/%d\n", n, len(sorted)) + } + } + } + + for w := 0; w < jobs; w++ { + wg.Add(1) + go worker() + } + for i := range sorted { + work <- i + } + close(work) + wg.Wait() + + if firstErr != nil { + return SweepResult{}, firstErr + } + + var res SweepResult + var pending []probeOutcome + for _, o := range outcomes { + switch o.state { + case Present: + res.Present = append(res.Present, o.sha) + case Absent: + res.Missing = append(res.Missing, o.sha) + default: + pending = append(pending, o) + } + } + + if cfg.ConfirmMax > 0 && len(pending) > cfg.ConfirmMax { + for _, o := range pending { + res.Unconfirmed = append(res.Unconfirmed, o.sha) + } + return res, nil + } + + for _, o := range pending { + state, err := confirm(ctx, b, o.sha, cfg.ConfirmRetries) + if err != nil { + return SweepResult{}, err + } + switch state { + case Present: + res.Present = append(res.Present, o.sha) + case Absent: + res.Missing = append(res.Missing, o.sha) + default: + res.Unconfirmed = append(res.Unconfirmed, o.sha) + } + } + + sort.Strings(res.Present) + sort.Strings(res.Missing) + sort.Strings(res.Unconfirmed) + + if len(sorted)%1000 != 0 { + fmt.Fprintf(progress, "probed %d/%d\n", len(sorted), len(sorted)) + } + + return res, nil +} + +// confirm re-probes sha serially up to retries attempts with linear backoff +// (1s, 2s, ... between attempts). A 2xx result confirms Present, a 404/410 +// confirms Missing (Absent) immediately. If it is still transient after all +// attempts, the state is left Unconfirmed rather than guessed as Missing. +func confirm(ctx context.Context, b Backend, sha string, retries int) (ProbeState, error) { + if retries < 1 { + retries = 1 + } + var last ProbeState = Unconfirmed + for attempt := 1; attempt <= retries; attempt++ { + state, transient, err := b.Probe(ctx, sha) + if err != nil && !transient { + return Unconfirmed, fmt.Errorf("confirm %s: %w", sha, err) + } + if state == Present || state == Absent { + return state, nil + } + last = Unconfirmed + if attempt < retries { + confirmSleep(time.Duration(attempt) * time.Second) + } + } + return last, nil +} diff --git a/internal/depot/sweep_test.go b/internal/depot/sweep_test.go new file mode 100644 index 0000000..78fdc1e --- /dev/null +++ b/internal/depot/sweep_test.go @@ -0,0 +1,169 @@ +package depot + +import ( + "context" + "io" + "sort" + "sync/atomic" + "testing" + "time" +) + +// stubBackend lets tests script Probe behavior per call. +type stubBackend struct { + probe func(ctx context.Context, sha string) (ProbeState, bool, error) +} + +func (s *stubBackend) Name() string { return "stub" } +func (s *stubBackend) Probe(ctx context.Context, sha string) (ProbeState, bool, error) { + return s.probe(ctx, sha) +} +func (s *stubBackend) Get(ctx context.Context, sha, dest string) error { return nil } +func (s *stubBackend) Put(ctx context.Context, sha, src string) error { return nil } + +func sweepTestCfg() Config { + return Config{ProbeJobs: 4, ConfirmRetries: 3, ConfirmMax: 200} +} + +func TestSweepAllPresent(t *testing.T) { + b := &stubBackend{probe: func(ctx context.Context, sha string) (ProbeState, bool, error) { + return Present, false, nil + }} + shas := []string{"bbb", "aaa", "ccc"} + res, err := Sweep(context.Background(), b, shas, sweepTestCfg(), io.Discard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + want := []string{"aaa", "bbb", "ccc"} + sort.Strings(want) + if len(res.Present) != 3 || len(res.Missing) != 0 || len(res.Unconfirmed) != 0 { + t.Fatalf("got %+v", res) + } + for i, sha := range res.Present { + if sha != want[i] { + t.Fatalf("expected sorted output, got %v", res.Present) + } + } +} + +func TestSweepMissing(t *testing.T) { + b := &stubBackend{probe: func(ctx context.Context, sha string) (ProbeState, bool, error) { + return Absent, false, nil + }} + shas := []string{"aaa"} + res, err := Sweep(context.Background(), b, shas, sweepTestCfg(), io.Discard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(res.Missing) != 1 || res.Missing[0] != "aaa" { + t.Fatalf("got %+v", res) + } + if len(res.Present) != 0 || len(res.Unconfirmed) != 0 { + t.Fatalf("got %+v", res) + } +} + +func TestSweepThrottledUnconfirmed(t *testing.T) { + var sleeps []time.Duration + orig := confirmSleep + confirmSleep = func(d time.Duration) { sleeps = append(sleeps, d) } + defer func() { confirmSleep = orig }() + + b := &stubBackend{probe: func(ctx context.Context, sha string) (ProbeState, bool, error) { + return Unconfirmed, true, nil + }} + shas := []string{"aaa"} + cfg := sweepTestCfg() + res, err := Sweep(context.Background(), b, shas, cfg, io.Discard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(res.Unconfirmed) != 1 || res.Unconfirmed[0] != "aaa" { + t.Fatalf("got %+v", res) + } + if len(res.Missing) != 0 { + t.Fatalf("must never collapse into missing: %+v", res) + } + if len(sleeps) != 2 || sleeps[0] != 1*time.Second || sleeps[1] != 2*time.Second { + t.Fatalf("expected sleeps [1s 2s], got %v", sleeps) + } +} + +func TestSweepTransientRecovers(t *testing.T) { + orig := confirmSleep + confirmSleep = func(d time.Duration) {} + defer func() { confirmSleep = orig }() + + var calls int32 + b := &stubBackend{probe: func(ctx context.Context, sha string) (ProbeState, bool, error) { + n := atomic.AddInt32(&calls, 1) + if n == 1 { + return Unconfirmed, true, nil // initial parallel probe: 428 + } + return Present, false, nil // serial re-probe: 206 + }} + res, err := Sweep(context.Background(), b, []string{"aaa"}, sweepTestCfg(), io.Discard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(res.Present) != 1 || res.Present[0] != "aaa" { + t.Fatalf("got %+v", res) + } +} + +func TestSweepConfirmMaxCap(t *testing.T) { + orig := confirmSleep + var sleepCalls int32 + confirmSleep = func(d time.Duration) { atomic.AddInt32(&sleepCalls, 1) } + defer func() { confirmSleep = orig }() + + b := &stubBackend{probe: func(ctx context.Context, sha string) (ProbeState, bool, error) { + return Unconfirmed, true, nil + }} + shas := []string{"a", "b", "c", "d", "e"} + cfg := sweepTestCfg() + cfg.ConfirmMax = 2 + res, err := Sweep(context.Background(), b, shas, cfg, io.Discard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(res.Unconfirmed) != 5 { + t.Fatalf("expected all 5 unconfirmed, got %+v", res) + } + if len(res.Missing) != 0 || len(res.Present) != 0 { + t.Fatalf("got %+v", res) + } + if sleepCalls != 0 { + t.Fatalf("expected zero serial re-probes/sleeps when over ConfirmMax, got %d calls", sleepCalls) + } +} + +func TestSweepParallelBounded(t *testing.T) { + var inFlight int32 + var maxInFlight int32 + shas := make([]string, 100) + for i := range shas { + shas[i] = string(rune('a'+i%26)) + string(rune('A'+i/26)) + } + b := &stubBackend{probe: func(ctx context.Context, sha string) (ProbeState, bool, error) { + n := atomic.AddInt32(&inFlight, 1) + for { + cur := atomic.LoadInt32(&maxInFlight) + if n <= cur || atomic.CompareAndSwapInt32(&maxInFlight, cur, n) { + break + } + } + time.Sleep(time.Millisecond) + atomic.AddInt32(&inFlight, -1) + return Present, false, nil + }} + cfg := sweepTestCfg() + cfg.ProbeJobs = 4 + _, err := Sweep(context.Background(), b, shas, cfg, io.Discard) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if maxInFlight > 4 { + t.Fatalf("expected max concurrency <= 4, got %d", maxInFlight) + } +} diff --git a/internal/dispatch/dispatch.go b/internal/dispatch/dispatch.go index 4b69e70..b83a93e 100644 --- a/internal/dispatch/dispatch.go +++ b/internal/dispatch/dispatch.go @@ -5,8 +5,10 @@ // Cutover state (Phase 5 → 6): the internal pipeline/topdata/erf/wiki // packages migrated from gitea/sow-tools now live in this tree, so wired // builders delegate to internal/app's legacy command surface (mapped per -// docs/command-surface.md). A builder with no migrated logic yet (depot) keeps -// the Wired=false fail-closed path: exit 70, never a faked artifact. +// docs/command-surface.md). depot is wired but bypasses that legacy surface +// entirely, delegating straight to internal/depot.Run. A builder with no +// migrated logic yet keeps the Wired=false fail-closed path: exit 70, never a +// faked artifact. package dispatch import ( @@ -16,6 +18,7 @@ import ( "git.westgate.pw/ShadowsOverWestgate/sow-tools/internal/app" "git.westgate.pw/ShadowsOverWestgate/sow-tools/internal/buildinfo" + "git.westgate.pw/ShadowsOverWestgate/sow-tools/internal/depot" "git.westgate.pw/ShadowsOverWestgate/sow-tools/internal/menu" ) @@ -90,8 +93,15 @@ var Registry = []Builder{ { Name: "depot", Bin: "crucible-depot", - Summary: "verify/move content-addressed depot blobs (SeaweedFS)", - Wired: false, // no migrated logic yet; fails closed (exit 70) + Summary: "content-addressed asset depot (local/cdn/bunny)", + Commands: []Command{ + {Name: "status", Summary: "report referenced-vs-present drift against a backend", Usage: "crucible depot status [--manifests DIR] [--source DIR] --target bunny|cdn|local"}, + {Name: "push", Summary: "upload referenced-but-absent blobs from a local depot", Usage: "crucible depot push [--manifests DIR] --source DIR --target bunny"}, + {Name: "verify", Summary: "existence sweep plus sampled download re-hash", Usage: "crucible depot verify [--manifests DIR] --target bunny|cdn [--sample N]"}, + {Name: "get", Summary: "fetch one blob with sha re-verify", Usage: "crucible depot get --target cdn|bunny|local"}, + {Name: "pull", Summary: "incremental verified pull of every referenced blob", Usage: "crucible depot pull [--manifests DIR] --dest DIR --target cdn|bunny"}, + }, + Wired: true, }, { Name: "hak", @@ -365,6 +375,12 @@ func runBuilder(name string, args []string, out, errw io.Writer) int { return exitOK } } + if b.Name == "depot" && b.Wired { + // depot parses its own subcommand (status/push/verify/get/pull) and + // has its own richer exit contract (0/1/2/64/70), so it bypasses the + // b.Commands/delegateLegacy routing entirely. + return depot.Run(args, out, errw, os.Getenv) + } if !b.Wired { // No migrated logic yet (depot): fail closed, never fake an artifact. fmt.Fprintf(errw, unwiredMsg, b.Name, b.Bin) diff --git a/internal/dispatch/dispatch_test.go b/internal/dispatch/dispatch_test.go index 5810b32..b7c4e7e 100644 --- a/internal/dispatch/dispatch_test.go +++ b/internal/dispatch/dispatch_test.go @@ -150,7 +150,7 @@ func TestBuilderHelpIsOK(t *testing.T) { func TestCanonicalCommandSurface(t *testing.T) { want := map[string][]string{ - "depot": nil, + "depot": {"status", "push", "verify", "get", "pull"}, "hak": {"build", "manifest"}, "module": {"build", "extract", "validate", "compare", "manifest"}, "topdata": {"validate", "build", "package", "compare", "convert"}, @@ -172,7 +172,11 @@ func TestRegistryCommandNamesAndAliasesAreUnambiguous(t *testing.T) { for _, builder := range Registry { seen := map[string]bool{} for _, command := range builder.Commands { - if command.Name == "" || command.Summary == "" || command.AppCommand == "" || command.Usage == "" { + // depot parses its own subcommands and bypasses AppCommand routing + // entirely (see the depot special-case in runBuilder), so its + // Commands carry no AppCommand. + requireAppCommand := builder.Name != "depot" + if command.Name == "" || command.Summary == "" || command.Usage == "" || (requireAppCommand && command.AppCommand == "") { t.Errorf("%s has incomplete command metadata: %#v", builder.Name, command) } if seen[command.Name] { @@ -289,15 +293,35 @@ func TestMenuItemsSkipsUnwiredIncludesWired(t *testing.T) { t.Fatal("expected menu items") } sawModuleBuild := false + sawDepotStatus := false for _, it := range items { - if it.Args[0] == "depot" { - t.Errorf("unwired builder %q must not appear in the menu", it.Args[0]) - } if len(it.Args) == 2 && it.Args[0] == "module" && it.Args[1] == "build" { sawModuleBuild = true } + if len(it.Args) == 2 && it.Args[0] == "depot" && it.Args[1] == "status" { + sawDepotStatus = true + } } if !sawModuleBuild { t.Error("expected 'module build' in the menu") } + if !sawDepotStatus { + t.Error("expected wired builder 'depot status' in the menu") + } +} + +// TestDepotRoutesToDepotRun asserts the depot special-case in runBuilder +// reaches depot.Run rather than the generic Commands/delegateLegacy path or +// the unwired fail-closed path. With no manifests dir present, depot.Run's +// status command fails resolving the backend/manifest (exit 70), but the +// stderr text must come from depot, never the dispatcher's unwiredMsg. +func TestDepotRoutesToDepotRun(t *testing.T) { + var out, errw bytes.Buffer + code := runBuilder("depot", []string{"status", "--target", "local"}, &out, &errw) + if errw.Len() == 0 { + t.Fatalf("depot status with no manifests dir: expected an error from depot.Run, got no stderr (exit=%d)", code) + } + if strings.Contains(errw.String(), "not wired yet") { + t.Fatalf("depot status: stderr shows the dispatcher's unwired message, want depot.Run's own error:\n%s", errw.String()) + } } diff --git a/scripts/crucible-smoke.sh b/scripts/crucible-smoke.sh index a497b71..44e8179 100755 --- a/scripts/crucible-smoke.sh +++ b/scripts/crucible-smoke.sh @@ -17,8 +17,8 @@ bin=bin "${bin}/crucible" list >/dev/null # Keep in sync with internal/dispatch.Registry (Wired flag). -unwired=(depot) -wired=(hak module topdata wiki) +unwired=() +wired=(depot hak module topdata wiki) exit_of() { set +e; "$@" >/dev/null 2>&1; local c=$?; set -e; echo "${c}"; }