diff --git a/CHANGELOG.md b/CHANGELOG.md index d4d2708..c232c4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Phase G (ops) — **cloud storage backends** (first G cycle): + - New `internal/storage` leaf defining a backend-neutral `Store` interface (`Put`/`Get`/`Delete`/`List`/`Stat`, all context-aware) with two implementations: **local** (the previous filesystem layout, now behind the interface — temp-write+rename for atomic publish, keys map verbatim to file names so existing catalogs need no migration) and **s3** (S3 and S3-compatible services like MinIO/R2 via AWS SDK v2; streaming transfer-manager upload that is atomic on completion, path-style for custom endpoints). + - `internal/dumps.Catalog` refactored to hold a `Store` and address dumps by key (`.dump`, `.meta.json`) instead of filesystem paths; `Path`/`MetaPath`/`Root` removed. Catalog methods (`PutDump`/`OpenDump`/`ReadMeta`/`WriteMeta`/`List`/`Delete`/`ResolveChain`/`PruneDryRun`) are now context-aware. `backup`, `restore`, `verify`, and `dumps` stream through the store; SHA-256 integrity is computed app-side over `envelope ++ body`, so a dump verifies identically whether it lives locally or in S3. + - New `storage:` config block (`type: local|s3`, `bucket`, `prefix`, `region`, `endpoint`) with fail-fast validation; S3 credentials resolve from the standard AWS chain (never stored in config). See [docs/STORAGE.md](docs/STORAGE.md). + - A shared `RunStoreSuite` contract test (mirroring Phase D's `RunDriverSuite`) runs against every backend: the local backend runs it as a unit test, and the S3 backend runs it against MinIO via testcontainers under the `integration` tag, so the live object-store path executes in CI. GCS/Azure backends are a fast-follow. + - Phase A skeleton: Go module, Cobra CLI with placeholder subcommands, Bubble Tea TUI placeholder, Driver interface + registry, errs package, config stub, Makefile, golangci-lint with depguard, CI workflow. - Phase B Postgres walking skeleton: `backup`, `restore`, `sync`, `verify`, `inspect`, `dumps`, and `profile` working end-to-end against PostgreSQL (shelling out to `pg_dump`/`pg_restore`, pgx for inspect), SHA-256 dump checksums with sidecar metadata, named profiles with `env:` secret refs, and POSIX exit-code taxonomy. - Phase C interactive TUI dashboard: multi-panel Bubble Tea UI (profiles · dumps · jobs) with live job-progress subscription, backup/restore modal forms, an error overlay, and golden snapshot tests. diff --git a/README.md b/README.md index d421cc9..8c12ede 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ A single binary that turns the painful, error-prone sprawl of `pg_dump` → `pg_ --- > [!WARNING] -> **Pre-1.0 — active development.** Postgres, MySQL, and MariaDB backup/restore/sync/verify/inspect work end-to-end today (Phases B + E), and bare `siphon` opens an interactive multi-panel dashboard (Phase C). The driver layer is hardened with a shared cross-driver test harness, capability gating, and connection retry (Phase D). Advanced transfer is complete (Phase F): incremental backup/restore, cross-engine sync, and CDC continuous replication all work end-to-end. Ops features (cloud storage, secret backends, retention, telemetry) are on the [roadmap](#roadmap). APIs, flags, and the on-disk dump format may change before 1.0. Track progress via the milestone tags (`phase-a` … `phase-f`). +> **Pre-1.0 — active development.** Postgres, MySQL, and MariaDB backup/restore/sync/verify/inspect work end-to-end today (Phases B + E), and bare `siphon` opens an interactive multi-panel dashboard (Phase C). The driver layer is hardened with a shared cross-driver test harness, capability gating, and connection retry (Phase D). Advanced transfer is complete (Phase F): incremental backup/restore, cross-engine sync, and CDC continuous replication all work end-to-end. Ops features (Phase G) are underway — the dump catalog can now live in an S3 / S3-compatible bucket (`storage:` config block); secret backends, retention, and telemetry are still on the [roadmap](#roadmap). APIs, flags, and the on-disk dump format may change before 1.0. Track progress via the milestone tags (`phase-a` … `phase-f`). ## Table of contents @@ -55,7 +55,7 @@ A single binary that turns the painful, error-prone sprawl of `pg_dump` → `pg_ | **D** — Driver hardening | Shared cross-driver test harness (`RunDriverSuite`), capability-gating helper (`RequireCapability`), Postgres connection-probe retry, and a `docs/DRIVERS.md` contributor guide | ✅ Complete | | **E** — MySQL + MariaDB | Both drivers via a shared `_mysqlcommon` package (shared `Conn`, `mysqldump`/`mariadb-dump` backup, client-pipe restore), exercised by the Phase D `RunDriverSuite` harness | ✅ Complete | | **F** — Advanced transfer | All four advanced-transfer modes work end-to-end: bounded-buffer streaming sync; **incremental** backup/restore (`backup --incremental --base ` captures a bounded change set via Postgres logical decoding / MySQL-MariaDB binlog, `restore` replays the base→incremental chain, Postgres orphan-slot sweep); **cross-engine** sync (`sync --cross-engine` — typed `SchemaInspector` introspection → canonical type-mapping, e.g. Postgres → MySQL); and **CDC** (`siphon cdc` / `sync --continuous` — unbounded change streaming with snapshot→stream handoff, resumable, same- and cross-engine). Live DB paths are integration-tested in CI — see [docs/INCREMENTAL.md](docs/INCREMENTAL.md), [docs/CROSS_ENGINE.md](docs/CROSS_ENGINE.md), [docs/CDC.md](docs/CDC.md) | ✅ Complete | -| **G** — Ops features | Cloud storage, secret backends, profile groups + 2FA, team mode, audit log, retention, telemetry | ⏳ Planned | +| **G** — Ops features | **Cloud storage** ships first: the dump catalog can live in an S3 / S3-compatible bucket via a pluggable `storage.Store` backend (local + S3 today; GCS/Azure are a fast-follow), selected by a `storage:` config block, with SHA-256 integrity preserved end-to-end — see [docs/STORAGE.md](docs/STORAGE.md). Secret backends, profile groups + 2FA, team mode, audit log, retention, and telemetry follow in later G cycles. | 🟡 In progress | | **H** — Distribution | GoReleaser, Homebrew tap, Scoop bucket, install script, docs site | ⏳ Planned | ## Requirements @@ -168,6 +168,19 @@ profiles: Secret references (`env:VAR`) are resolved at runtime, so the config file is safe to commit. Plain values are also accepted. +By default the dump catalog lives on local disk at `defaults.dump_dir`. To store dumps in an S3 / S3-compatible bucket instead, add a `storage:` block: + +```yaml +storage: + type: s3 # "local" (default) | "s3" + bucket: my-siphon-dumps # required for s3 + prefix: prod # optional key prefix + region: us-east-1 + endpoint: "" # optional: custom endpoint for MinIO / R2 +``` + +Credentials are resolved from the standard AWS chain (env vars, `~/.aws`, instance role) — never stored in the config file. See [docs/STORAGE.md](docs/STORAGE.md) for details. + ## Architecture siphon is a strictly layered Go application; imports flow **downward only**, enforced at lint time by `golangci-lint`'s `depguard` (an upward import fails CI): diff --git a/docs/STORAGE.md b/docs/STORAGE.md new file mode 100644 index 0000000..cc005c5 --- /dev/null +++ b/docs/STORAGE.md @@ -0,0 +1,100 @@ +# Storage backends + +By default siphon keeps the dump catalog — every dump body plus its sidecar +metadata — in a local directory. Phase G adds a pluggable storage layer so the +catalog can instead live in an S3 (or S3-compatible) bucket, with `backup`, +`restore`, `verify`, and `dumps` all reading and writing through it transparently. + +## Table of contents + +- [How it works](#how-it-works) +- [Configuration](#configuration) +- [Integrity across backends](#integrity-across-backends) +- [Scope and limitations](#scope-and-limitations) + +## How it works + +The dump catalog (`internal/dumps`) holds a `storage.Store` and addresses +objects by opaque keys — `.dump` for the envelope-prefixed dump body and +`.meta.json` for the sidecar — never by filesystem path. The `Store` +interface is small and backend-neutral: + +```go +type Store interface { + Put(ctx, key, io.Reader) error // durable, atomic-on-complete + Get(ctx, key) (io.ReadCloser, error) // one-shot forward stream + Delete(ctx, key) error // idempotent + List(ctx) ([]string, error) + Stat(ctx, key) (size int64, exists bool, err error) +} +``` + +Two backends ship today (`internal/storage`): + +- **local** — a single directory. `Put` writes to a temp file and renames, so a + failed or cancelled write never leaves a partial dump under its final key. + Keys map verbatim to file names, so a pre-Phase-G local catalog keeps working + with no migration. +- **s3** — an S3 or S3-compatible bucket (AWS, MinIO, Cloudflare R2). `Put` + streams through the SDK's transfer manager, so the object only becomes + addressable once the upload completes — the same atomic-on-complete guarantee + the local backend gets from rename. + +A backup stages the dump body to a local temp file (the dump tool needs a real +fd), then streams `envelope ++ body` into the store in a single `Put`, teeing +through SHA-256 as the bytes flow. Restore and verify open the dump with `Get` +and stream it straight into the envelope reader — no full local download. + +## Configuration + +Storage is selected by a top-level `storage:` block in the config file. Omitting +it (or `type: local`) uses the local filesystem at `defaults.dump_dir`. + +```yaml +version: 1 +defaults: + dump_dir: ~/.local/share/siphon/dumps # used by the local backend + +storage: + type: s3 # "local" (default) | "s3" + bucket: my-siphon-dumps # required for s3 + prefix: prod # optional key prefix within the bucket + region: us-east-1 + endpoint: "" # optional: custom endpoint for S3-compatible services +``` + +For an S3-compatible service such as MinIO or R2, set `endpoint` to its URL +(path-style addressing is used automatically). + +**Credentials are never stored in the config file.** The S3 backend resolves +them from the standard AWS chain — `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY` +environment variables, the shared `~/.aws/config`, or an instance/role profile — +so the config stays safe to commit. + +## Integrity across backends + +The SHA-256 checksum is computed over the `envelope ++ body` stream at write +time and recomputed over the `Get` stream at verify time — both in siphon, not +in the backend. A dump written to S3 and a dump written locally therefore verify +identically, and `siphon verify` catches corruption regardless of where the dump +lives. siphon does not trust the backend's own ETag/MD5 (multipart uploads do +not expose a plain object MD5). + +The live S3 path is integration-tested in CI against MinIO via testcontainers +(`internal/storage/s3_integration_test.go`) using the same `RunStoreSuite` +contract that the local backend runs, so the streaming upload, ranged read, +listing, and not-found mapping all execute against a real object store — not +just compile. + +## Scope and limitations + +- Backends covered: **local** and **s3** (incl. S3-compatible). **GCS and Azure + Blob are not implemented yet** — they are a fast-follow, and will get the full + correctness bar for free by running the same `RunStoreSuite` contract. +- `dumps list` over S3 issues one `ListObjectsV2` plus one read per metadata + object (N+1). This is fine at expected catalog sizes; no pagination/parallel + optimization is done yet. +- Each `Get` is a fresh one-shot forward stream — callers must not assume the + returned reader is seekable. +- Retention/lifecycle (chain-aware pruning over remote storage) remains a + separate Phase G concern; the `Store.Delete` it needs is delivered here. diff --git a/go.mod b/go.mod index d092437..3644eab 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,12 @@ module github.com/nixrajput/siphon go 1.26.3 require ( + github.com/aws/aws-sdk-go-v2 v1.42.0 + github.com/aws/aws-sdk-go-v2/config v1.32.25 + github.com/aws/aws-sdk-go-v2/credentials v1.19.24 + github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.2.11 + github.com/aws/aws-sdk-go-v2/service/s3 v1.104.0 + github.com/aws/smithy-go v1.27.1 github.com/charmbracelet/bubbles v1.0.0 github.com/charmbracelet/bubbletea v1.3.10 github.com/charmbracelet/huh v1.0.0 @@ -16,6 +22,7 @@ require ( github.com/spf13/cobra v1.10.2 github.com/testcontainers/testcontainers-go v0.42.0 github.com/testcontainers/testcontainers-go/modules/mariadb v0.42.0 + github.com/testcontainers/testcontainers-go/modules/minio v0.42.0 github.com/testcontainers/testcontainers-go/modules/mysql v0.42.0 github.com/testcontainers/testcontainers-go/modules/postgres v0.42.0 gopkg.in/yaml.v3 v3.0.1 @@ -27,6 +34,19 @@ require ( github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect github.com/Microsoft/go-winio v0.6.2 // indirect github.com/atotto/clipboard v0.1.4 // indirect + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.13 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.29 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.29 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.29 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.30 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.22 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.29 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.29 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.2.0 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.31.3 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.6 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.43.3 // indirect github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect github.com/aymanbagabas/go-udiff v0.3.1 // indirect github.com/catppuccin/go v0.3.0 // indirect diff --git a/go.sum b/go.sum index da9bbc7..0753714 100644 --- a/go.sum +++ b/go.sum @@ -12,6 +12,44 @@ github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERo github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/atotto/clipboard v0.1.4 h1:EH0zSVneZPSuFR11BlR9YppQTVDbh5+16AmcJi4g1z4= github.com/atotto/clipboard v0.1.4/go.mod h1:ZY9tmq7sm5xIbd9bOK4onWV4S6X0u6GY7Vn0Yu86PYI= +github.com/aws/aws-sdk-go-v2 v1.42.0 h1:XvXMJTkFQtpBKIWZnmr9ZEOc2InWM2yldjXEJ/bymhA= +github.com/aws/aws-sdk-go-v2 v1.42.0/go.mod h1:27+ACypSLljLAEKsCYOmrjKh83vuTRkuAe9Uv/3A4bg= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.13 h1:p1BBrg/Hhp6uK7zpejeI8QFXHJeC/mynzi04Sl03k9g= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.13/go.mod h1:8cIfkE9MDhkRZGpQ22aV6/lkYeYSozpz16Smrs5x4Ls= +github.com/aws/aws-sdk-go-v2/config v1.32.25 h1:ACCejvStYoilgwrfegSt5ZntCbPrk52qfwyNcnl3omM= +github.com/aws/aws-sdk-go-v2/config v1.32.25/go.mod h1:LJyU8sDRbXUxFn8xMJIGP+v9QYYwveNLI8a/giAOiAs= +github.com/aws/aws-sdk-go-v2/credentials v1.19.24 h1:2hQqYCV9yqyePQ9o6dCrZc/zO8U3TwPr9mIKlZnPu/I= +github.com/aws/aws-sdk-go-v2/credentials v1.19.24/go.mod h1:IDwpACtwqHLISdzfwUUNq4P9DsB/h5BLg4FwJPNfqFY= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.29 h1:r6qZHbT+wxgWO/e9vYNUEtg7lv5+UN3pRqKhLXvnArg= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.29/go.mod h1:QRnaRcTVGKPGRy8w78HMQtKUGRYcnMZAANATkeVA6Mo= +github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.2.11 h1:ORrr68X2RQ4OBNI4aYJapjcmbaG5mst0Z7/tjUboZEA= +github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager v0.2.11/go.mod h1:W4kVdkT0M/UDMNncotFG2X/HGX69yMdNn7S9WGMhUjk= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.29 h1:f3vKqSo13fhTYb+JEcXwXefZQE26I1FB5eTSniU67ko= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.29/go.mod h1:MzoLFUArKGpGD+ukmPiTPG1X5x4o6M2kq4v2dr1FiEc= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.29 h1:RdwIf/CuUsvJX3RgJagbOyotl/cxoLY4xviKuE7p2GY= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.29/go.mod h1:71wt8W2EgswdZy9Mf9KNnzxZ3TiZlv4caKghPktDOkA= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.30 h1:VTGy885W5DKBxWRUJbym9hytNaYzsyaPkCHGRRMAOhU= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.30/go.mod h1:AS0HycUvJRFvTt613AYDOgO2jzw+00cVSMny8XB3yMY= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12 h1:ZD2+BSw9vFsNlKYIasSNt3uDbjqqXIBcM13UJv/Lx2k= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12/go.mod h1:Ms4zlcVBbXbiP7EVLhl+lgjvA/a7YphqQ3Ih3174EmI= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.22 h1:V51LGlOq/1VsDsHUdoklAQi7rMmx4qQubvFYAlP2254= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.22/go.mod h1:4Pzhyz8hJOm2bepgl+NjvRx8vlUFAIIvJnZ/MkcNPpU= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.29 h1:DRebniUGZ2MqiiIVmQJ04vIXr918hubdHMnarSLEWyU= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.29/go.mod h1:LfRkPCD8YHDM2E5eTkos2UpwYeZnBcVarTa8L59bJHA= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.29 h1:hiME6pBzC7OTl9LMtlyTWBuEl1f4QBcUmFDKC7MLXtc= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.29/go.mod h1:G7RP+uhagpKtKhd1BM9N6JQqjCcGEU47K5lBVZQyRQw= +github.com/aws/aws-sdk-go-v2/service/s3 v1.104.0 h1:ta8csKy5vN91F3i5gGR85lFV0srBqySEji7Jroes6rE= +github.com/aws/aws-sdk-go-v2/service/s3 v1.104.0/go.mod h1:77ZAgynvx1txMvDG8gGWoWkO1augYDxkp9JElWFgjQU= +github.com/aws/aws-sdk-go-v2/service/signin v1.2.0 h1:3nXpRcFwRCW8n7HgO2QGy0Dc20eQNfBuUemGQhpF8m8= +github.com/aws/aws-sdk-go-v2/service/signin v1.2.0/go.mod h1:LxYujSTLPRlp2vTtcUO/+1ilrew8ytt6SvQyOgejzFQ= +github.com/aws/aws-sdk-go-v2/service/sso v1.31.3 h1:ey1XLTYXb9PcLt4535632o5kCGXNXEhNb620Dqwuylo= +github.com/aws/aws-sdk-go-v2/service/sso v1.31.3/go.mod h1:Lk7PlmoTYryQmyBG0EXqj5BcUbj3whXdU2s3yGI3EAc= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.6 h1:yLr03zQE/5Eu5l3QU0Si+xMbLMbSDF2YXsigqXngs6g= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.6/go.mod h1:Q5N6icH+KJZDLh+ESNwzdv6cZ6vLFF/egy3IOxWhmz4= +github.com/aws/aws-sdk-go-v2/service/sts v1.43.3 h1:VrIhKRCSK1umelSgB9RghvA9RTUYeQffyAS5ApXehNI= +github.com/aws/aws-sdk-go-v2/service/sts v1.43.3/go.mod h1:r8wkDOuLaaMFqFiYAb8dGY2A3gJCOujMc6CFOVC4Zhc= +github.com/aws/smithy-go v1.27.1 h1:4T340VFndXtADGF52gYa1POyL7s9E4Z1OeZ1hCscIw8= +github.com/aws/smithy-go v1.27.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k= github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8= github.com/aymanbagabas/go-udiff v0.3.1 h1:LV+qyBQ2pqe0u42ZsUEtPiCaUoqgA9gYRDs3vj1nolY= @@ -118,8 +156,12 @@ github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw= github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/cpuid/v2 v2.2.6 h1:ndNyv040zDGIDh8thGkXYjnFtiN02M1PVVF+JE/48xc= +github.com/klauspost/cpuid/v2 v2.2.6/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= @@ -142,6 +184,12 @@ github.com/mattn/go-runewidth v0.0.19 h1:v++JhqYnZuu5jSKrk9RbgF5v4CGUjqRfBm05byF github.com/mattn/go-runewidth v0.0.19/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs= github.com/mdelapenya/tlscert v0.2.0 h1:7H81W6Z/4weDvZBNOfQte5GpIMo0lGYEeWbkGp5LJHI= github.com/mdelapenya/tlscert v0.2.0/go.mod h1:O4njj3ELLnJjGdkN7M/vIVCpZ+Cf0L6muqOG4tLSl8o= +github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34= +github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM= +github.com/minio/minio-go/v7 v7.0.68 h1:hTqSIfLlpXaKuNy4baAp4Jjy2sqZEN9hRxD0M4aOfrQ= +github.com/minio/minio-go/v7 v7.0.68/go.mod h1:XAvOPJQ5Xlzk5o3o/ArO2NMbhSGkimC+bpW/ngRKDmQ= +github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM= +github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8= github.com/mitchellh/hashstructure/v2 v2.0.2 h1:vGKWl0YJqUNxE8d+h8f6NJLcCJrgbhC4NcD46KavDd4= github.com/mitchellh/hashstructure/v2 v2.0.2/go.mod h1:MG3aRVU/N29oo/V/IhBX8GR/zz4kQkprJgF2EVszyDE= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= @@ -162,6 +210,10 @@ github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28= github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 h1:ZK8zHtRHOkbHy6Mmr5D264iyp3TiX5OmNcI5cIARiQI= github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6/go.mod h1:CJlz5H+gyd6CUWT45Oy4q24RdLyn7Md9Vj2/ldJBSIo= github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELUXHmA= @@ -183,6 +235,8 @@ github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ= github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88= github.com/rogpeppe/go-internal v1.15.0 h1:D0RCU5rMAp+SpgkiNdrjfJ+LX4J1M32V2NeCY7EJ6hc= github.com/rogpeppe/go-internal v1.15.0/go.mod h1:DrUVZyrJU+txYW5/1kwtXQSMFio52ZOxX7yM1VHvnxs= +github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc= +github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/sahilm/fuzzy v0.1.1 h1:ceu5RHF8DGgoi+/dR5PsECjCDH1BE3Fnmpo7aVXOdRA= github.com/sahilm/fuzzy v0.1.1/go.mod h1:VFvziUEIMCrT6A6tw2RFIXPXXmzXbOsSHF0DOI8ZK9Y= @@ -205,6 +259,8 @@ github.com/testcontainers/testcontainers-go v0.42.0 h1:He3IhTzTZOygSXLJPMX7n44Xt github.com/testcontainers/testcontainers-go v0.42.0/go.mod h1:vZjdY1YmUA1qEForxOIOazfsrdyORJAbhi0bp8plN30= github.com/testcontainers/testcontainers-go/modules/mariadb v0.42.0 h1:ZfWUJSIDnbNgoLAXMV1fc7lqcxBIX3zdnhwjaVUo7N0= github.com/testcontainers/testcontainers-go/modules/mariadb v0.42.0/go.mod h1:0kV+yHee7zAgp0yccydxjNnHvlC1EOavTLCeg/lnRBY= +github.com/testcontainers/testcontainers-go/modules/minio v0.42.0 h1:8yTWNv8ALG7JQHYvm1n9PegH0uJT7dRtWNHf6eQeTRs= +github.com/testcontainers/testcontainers-go/modules/minio v0.42.0/go.mod h1:bcjonmVMA/aEzxFFIh/FRwSkeZ+fnxwvkGN/Z4EiW28= github.com/testcontainers/testcontainers-go/modules/mysql v0.42.0 h1:Yhv1k7vDpyzZePntg5R5Oj4ZMCyWpAfpJeRu1ROsgiU= github.com/testcontainers/testcontainers-go/modules/mysql v0.42.0/go.mod h1:Z7SCTuiZlghAdRjkv3Ir0iXJKC2T2avbtxLR0DRe+ng= github.com/testcontainers/testcontainers-go/modules/postgres v0.42.0 h1:GCbb1ndrF7OTDiIvxXyItaDab4qkzTFJ48LKFdM7EIo= @@ -236,6 +292,8 @@ golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI= golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo= +golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o= +golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -253,6 +311,8 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA= +gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/app/backup.go b/internal/app/backup.go index 8761972..8a36aa1 100644 --- a/internal/app/backup.go +++ b/internal/app/backup.go @@ -3,12 +3,12 @@ package app import ( + "bytes" "context" "crypto/sha256" "encoding/hex" "io" "os" - "path/filepath" "time" "github.com/oklog/ulid/v2" @@ -94,7 +94,7 @@ func Backup(parent context.Context, d Deps, opt BackupOpts) (<-chan jobs.Event, // NOTHING / INSERT IGNORE) so such a re-apply is harmless. Capturing // after (not before) the dump avoids the inverse hazard of streaming // pre-snapshot inserts that would conflict on PK. - bodyTmp, err := os.CreateTemp(d.Dumps.Root(), "siphon-base-body-*") + bodyTmp, err := os.CreateTemp("", "siphon-base-body-*") if err != nil { return err } @@ -138,15 +138,6 @@ func Backup(parent context.Context, d Deps, opt BackupOpts) (<-chan jobs.Event, } id := ulid.Make().String() - tmpPath := filepath.Join(d.Dumps.Root(), id+".dump.tmp") - finalPath := d.Dumps.Path(id) - - f, err := os.Create(tmpPath) - if err != nil { - return err - } - h := sha256.New() - tee := io.MultiWriter(f, h) env := &dumps.Envelope{ Type: dumps.EnvelopeBase, @@ -158,60 +149,36 @@ func Backup(parent context.Context, d Deps, opt BackupOpts) (<-chan jobs.Event, env.BinlogFile = basePos.BinlogFile env.BinlogEnd = basePos.BinlogPos } - if _, err := dumps.WriteEnvelope(tee, env); err != nil { - _ = f.Close() - _ = os.Remove(tmpPath) - return err - } - body, err := os.Open(bodyPath) + // Assemble the dump as envelope ++ body and stream it through the + // catalog's store under id, teeing through sha256 as it flows. The + // store publishes the object atomically, so a failed/cancelled Put + // leaves nothing addressable by id. Hashing stays here (app-side), so + // integrity is identical regardless of the storage backend. + size, checksum, err := putDump(ctx, d.Dumps, id, env, bodyPath) if err != nil { - _ = f.Close() - _ = os.Remove(tmpPath) - return err - } - if _, err := io.Copy(tee, body); err != nil { - _ = body.Close() - _ = f.Close() - _ = os.Remove(tmpPath) - return err - } - _ = body.Close() - - closeErr := f.Close() - if closeErr != nil { - _ = os.Remove(tmpPath) - return &errs.Error{Op: "app.backup", Code: errs.CodeSystem, Cause: closeErr, Hint: "failed to flush dump to disk (out of space?)"} - } - - if err := os.Rename(tmpPath, finalPath); err != nil { - _ = os.Remove(tmpPath) return err } - st, _ := os.Stat(finalPath) - size := int64(0) - if st != nil { - size = st.Size() - } - meta := &dumps.Meta{ ID: id, Profile: opt.Profile, Driver: resolved.Driver, SizeBytes: size, - Checksum: "sha256:" + hex.EncodeToString(h.Sum(nil)), + Checksum: checksum, Created: time.Now(), DumpFormat: "custom", } - if writeErr := d.Dumps.WriteMeta(meta); writeErr != nil { - // The catalog enumerates by sidecar metadata, so a dump without - // its meta would be an invisible orphan that never gets pruned. - _ = os.Remove(finalPath) + // Write meta LAST: the catalog enumerates by sidecar metadata, so a + // dump body without its meta is an invisible (prunable) orphan, never + // a dangling catalog entry. On meta failure, best-effort remove the + // orphaned body. + if writeErr := d.Dumps.WriteMeta(ctx, meta); writeErr != nil { + _ = d.Dumps.Delete(ctx, id) return writeErr } - emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "wrote " + finalPath}) + emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "wrote dump " + id}) return nil }, }) @@ -227,7 +194,7 @@ func Backup(parent context.Context, d Deps, opt BackupOpts) (<-chan jobs.Event, // envelope(end position) followed by the temp body — checksumming the whole file // as it is written, mirroring the full-backup tmp→rename + sha + WriteMeta flow. func runIncrementalBackup(ctx context.Context, d Deps, opt BackupOpts, resolved driver.Profile, conn driver.Conn, emit func(jobs.Event)) error { - base, err := d.Dumps.ReadMeta(opt.BaseID) + base, err := d.Dumps.ReadMeta(ctx, opt.BaseID) if err != nil { return &errs.Error{ Op: "app.backup.incremental", @@ -249,7 +216,7 @@ func runIncrementalBackup(ctx context.Context, d Deps, opt BackupOpts, resolved } } - since, err := basePosition(d, opt.BaseID) + since, err := basePosition(ctx, d, opt.BaseID) if err != nil { return err } @@ -274,9 +241,11 @@ func runIncrementalBackup(ctx context.Context, d Deps, opt BackupOpts, resolved } } - // Stream the change body to a temp file so we learn the end Position before - // writing the envelope (which must carry that end Position). - bodyTmp, err := os.CreateTemp(d.Dumps.Root(), "siphon-inc-body-*") + // Stream the change body to a local temp file so we learn the end Position + // before writing the envelope (which must carry that end Position). The body + // is staged locally regardless of the catalog's storage backend, then + // published in one streamed Put. + bodyTmp, err := os.CreateTemp("", "siphon-inc-body-*") if err != nil { return err } @@ -293,17 +262,7 @@ func runIncrementalBackup(ctx context.Context, d Deps, opt BackupOpts, resolved return &errs.Error{Op: "app.backup.incremental", Code: errs.CodeSystem, Cause: closeErr, Hint: "failed to flush change body to disk"} } - // Assemble the final dump: envelope(end position) + change body. id := ulid.Make().String() - tmpPath := filepath.Join(d.Dumps.Root(), id+".dump.tmp") - finalPath := d.Dumps.Path(id) - - f, err := os.Create(tmpPath) - if err != nil { - return err - } - h := sha256.New() - tee := io.MultiWriter(f, h) // The chain root: if the base is itself an incremental, inherit its BaseID; // otherwise the base IS the root. @@ -323,71 +282,81 @@ func runIncrementalBackup(ctx context.Context, d Deps, opt BackupOpts, resolved env.BinlogFile = endPos.BinlogFile env.BinlogEnd = endPos.BinlogPos } - if _, err := dumps.WriteEnvelope(tee, env); err != nil { - _ = f.Close() - _ = os.Remove(tmpPath) - return err - } - body, err := os.Open(bodyPath) + size, checksum, err := putDump(ctx, d.Dumps, id, env, bodyPath) if err != nil { - _ = f.Close() - _ = os.Remove(tmpPath) - return err - } - if _, err := io.Copy(tee, body); err != nil { - _ = body.Close() - _ = f.Close() - _ = os.Remove(tmpPath) return err } - _ = body.Close() - closeErr = f.Close() - if closeErr != nil { - _ = os.Remove(tmpPath) - return &errs.Error{Op: "app.backup.incremental", Code: errs.CodeSystem, Cause: closeErr, Hint: "failed to flush dump to disk (out of space?)"} - } - if err := os.Rename(tmpPath, finalPath); err != nil { - _ = os.Remove(tmpPath) - return err - } - - st, _ := os.Stat(finalPath) - size := int64(0) - if st != nil { - size = st.Size() - } meta := &dumps.Meta{ ID: id, Profile: opt.Profile, Driver: resolved.Driver, SizeBytes: size, - Checksum: "sha256:" + hex.EncodeToString(h.Sum(nil)), + Checksum: checksum, Created: time.Now(), DumpFormat: "jsonl-changes", BaseID: root, ParentID: opt.BaseID, } - if writeErr := d.Dumps.WriteMeta(meta); writeErr != nil { - _ = os.Remove(finalPath) + if writeErr := d.Dumps.WriteMeta(ctx, meta); writeErr != nil { + _ = d.Dumps.Delete(ctx, id) return writeErr } - emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "wrote " + finalPath}) + emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "wrote dump " + id}) return nil } +// putDump assembles a dump as the 4 KB envelope followed by the staged body +// file at bodyPath, streams it into the catalog under id, and returns the total +// byte count and the sha256 checksum (computed over the same envelope++body +// stream). The store publishes atomically, so a failed Put leaves nothing +// addressable by id. Counting size and hashing inline keeps both backend- +// agnostic — neither depends on the object being a local file afterward. +func putDump(ctx context.Context, cat *dumps.Catalog, id string, env *dumps.Envelope, bodyPath string) (int64, string, error) { + var hdr bytes.Buffer + if _, err := dumps.WriteEnvelope(&hdr, env); err != nil { + return 0, "", err + } + body, err := os.Open(bodyPath) + if err != nil { + return 0, "", err + } + defer func() { _ = body.Close() }() + + h := sha256.New() + counter := &countWriter{} + // Reader order: envelope header bytes, then the body file. Tee through the + // hash and a byte counter as the store consumes the stream. + src := io.TeeReader(io.MultiReader(&hdr, body), io.MultiWriter(h, counter)) + + if err := cat.PutDump(ctx, id, src); err != nil { + return 0, "", err + } + return counter.n, "sha256:" + hex.EncodeToString(h.Sum(nil)), nil +} + +// countWriter is an io.Writer that only counts the bytes written to it, so the +// dump's size can be recorded as the assembled stream flows to the store +// (which may be remote, so os.Stat on the result is not an option). +type countWriter struct{ n int64 } + +func (c *countWriter) Write(p []byte) (int, error) { + c.n += int64(len(p)) + return len(p), nil +} + // basePosition reads the end Position recorded in the base dump's envelope. The // next incremental resumes from here: WALEnd for Postgres, BinlogFile+BinlogEnd // for MySQL/MariaDB. -func basePosition(d Deps, baseID string) (canonical.Position, error) { - f, err := os.Open(d.Dumps.Path(baseID)) +func basePosition(ctx context.Context, d Deps, baseID string) (canonical.Position, error) { + rc, err := d.Dumps.OpenDump(ctx, baseID) if err != nil { return canonical.Position{}, err } - defer func() { _ = f.Close() }() - env, _, err := dumps.ReadEnvelope(f) + defer func() { _ = rc.Close() }() + env, _, err := dumps.ReadEnvelope(rc) if err != nil { return canonical.Position{}, err } diff --git a/internal/app/backup_test.go b/internal/app/backup_test.go index cee255f..e613d34 100644 --- a/internal/app/backup_test.go +++ b/internal/app/backup_test.go @@ -76,7 +76,7 @@ func TestBackup_WritesDumpAndMeta(t *testing.T) { for range ch { /* drain */ } - got, err := cat.List() + got, err := cat.List(context.Background()) if err != nil { t.Fatal(err) } @@ -140,7 +140,7 @@ func TestBackup_FullStampsBasePosition(t *testing.T) { for range ch { /* drain */ } - got, err := cat.List() + got, err := cat.List(context.Background()) if err != nil { t.Fatal(err) } @@ -148,7 +148,7 @@ func TestBackup_FullStampsBasePosition(t *testing.T) { t.Fatalf("catalog.List() = %d entries; want 1", len(got)) } - pos, err := basePosition(deps, got[0].ID) + pos, err := basePosition(context.Background(), deps, got[0].ID) if err != nil { t.Fatalf("basePosition: %v", err) } @@ -206,7 +206,7 @@ func TestBackup_DumpError_CleansTmpAndWritesNoCatalogEntry(t *testing.T) { } // No catalog entry should have been recorded. - got, err := cat.List() + got, err := cat.List(context.Background()) if err != nil { t.Fatal(err) } diff --git a/internal/app/restore.go b/internal/app/restore.go index f4948a1..af885eb 100644 --- a/internal/app/restore.go +++ b/internal/app/restore.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "io" - "os" "github.com/nixrajput/siphon/internal/canonical" "github.com/nixrajput/siphon/internal/driver" @@ -38,7 +37,7 @@ func Restore(parent context.Context, d Deps, opt RestoreOpts) (<-chan jobs.Event return nil, "", err } - chain, err := d.Dumps.ResolveChain(opt.DumpID) + chain, err := d.Dumps.ResolveChain(parent, opt.DumpID) if err != nil { return nil, "", err } @@ -76,7 +75,7 @@ func Restore(parent context.Context, d Deps, opt RestoreOpts) (<-chan jobs.Event for i, m := range chain { emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "applying " + m.ID}) - f, err := os.Open(d.Dumps.Path(m.ID)) + f, err := d.Dumps.OpenDump(ctx, m.ID) if err != nil { return err } diff --git a/internal/app/restore_chain_test.go b/internal/app/restore_chain_test.go index 54d9605..83c4e9e 100644 --- a/internal/app/restore_chain_test.go +++ b/internal/app/restore_chain_test.go @@ -1,11 +1,11 @@ package app import ( + "bytes" "context" "encoding/json" "errors" "io" - "os" "testing" "time" @@ -87,21 +87,18 @@ func (d *chainFakeDriver) Connect(_ context.Context, _ driver.Profile) (driver.C // via conn.ApplyChange), so the change's table records the payload. func writeChainDump(t *testing.T, cat *dumps.Catalog, id, baseID, parentID, payload string) { t.Helper() - f, err := os.Create(cat.Path(id)) - if err != nil { - t.Fatalf("create dump %s: %v", id, err) - } + ctx := context.Background() typ := dumps.EnvelopeIncremental if parentID == "" { typ = dumps.EnvelopeBase } - if _, err := dumps.WriteEnvelope(f, &dumps.Envelope{ + var buf bytes.Buffer + if _, err := dumps.WriteEnvelope(&buf, &dumps.Envelope{ Type: typ, Driver: "fake", BaseID: baseID, ParentID: parentID, }); err != nil { - _ = f.Close() t.Fatalf("write envelope %s: %v", id, err) } body := payload @@ -114,14 +111,11 @@ func writeChainDump(t *testing.T, cat *dumps.Catalog, id, baseID, parentID, payl } body = string(js) + "\n" } - if _, err := io.WriteString(f, body); err != nil { - _ = f.Close() - t.Fatalf("write payload %s: %v", id, err) - } - if err := f.Close(); err != nil { - t.Fatalf("close dump %s: %v", id, err) + buf.WriteString(body) + if err := cat.PutDump(ctx, id, &buf); err != nil { + t.Fatalf("put dump %s: %v", id, err) } - if err := cat.WriteMeta(&dumps.Meta{ + if err := cat.WriteMeta(ctx, &dumps.Meta{ ID: id, Profile: "test", Driver: "fake", @@ -270,26 +264,20 @@ func TestRestoreChain_UpToUnknownErrors(t *testing.T) { // any destructive Clean runs. func writeMismatchDump(t *testing.T, cat *dumps.Catalog, id, envDriver string) { t.Helper() - f, err := os.Create(cat.Path(id)) - if err != nil { - t.Fatalf("create dump %s: %v", id, err) - } - if _, err := dumps.WriteEnvelope(f, &dumps.Envelope{ + ctx := context.Background() + var buf bytes.Buffer + if _, err := dumps.WriteEnvelope(&buf, &dumps.Envelope{ Type: dumps.EnvelopeBase, Driver: envDriver, BaseID: id, }); err != nil { - _ = f.Close() t.Fatalf("write envelope %s: %v", id, err) } - if _, err := io.WriteString(f, "payload"); err != nil { - _ = f.Close() - t.Fatalf("write payload %s: %v", id, err) - } - if err := f.Close(); err != nil { - t.Fatalf("close dump %s: %v", id, err) + buf.WriteString("payload") + if err := cat.PutDump(ctx, id, &buf); err != nil { + t.Fatalf("put dump %s: %v", id, err) } - if err := cat.WriteMeta(&dumps.Meta{ + if err := cat.WriteMeta(ctx, &dumps.Meta{ ID: id, Profile: "test", Driver: envDriver, diff --git a/internal/app/roundtrip_test.go b/internal/app/roundtrip_test.go index 3debe6d..77492f6 100644 --- a/internal/app/roundtrip_test.go +++ b/internal/app/roundtrip_test.go @@ -88,7 +88,7 @@ func TestBackupRestoreVerify_Roundtrip(t *testing.T) { } drain(t, bch) - entries, err := cat.List() + entries, err := cat.List(ctx) if err != nil { t.Fatal(err) } diff --git a/internal/app/verify.go b/internal/app/verify.go index b630eb5..7252a78 100644 --- a/internal/app/verify.go +++ b/internal/app/verify.go @@ -5,7 +5,6 @@ import ( "crypto/sha256" "encoding/hex" "io" - "os" "time" "github.com/nixrajput/siphon/internal/driver" @@ -16,13 +15,13 @@ import ( // dump file and comparing it against the checksum recorded in the meta sidecar. // It is stateless — no DB connection is required (Phase B checksums only; Phase // F adds envelope-header validation via the driver's Verify method). -func Verify(_ context.Context, d Deps, dumpID string) (*driver.VerifyReport, error) { - meta, err := d.Dumps.ReadMeta(dumpID) +func Verify(ctx context.Context, d Deps, dumpID string) (*driver.VerifyReport, error) { + meta, err := d.Dumps.ReadMeta(ctx, dumpID) if err != nil { return nil, err } - f, err := os.Open(d.Dumps.Path(dumpID)) + f, err := d.Dumps.OpenDump(ctx, dumpID) if err != nil { return nil, err } diff --git a/internal/cli/dumps.go b/internal/cli/dumps.go index 3302361..6d7d565 100644 --- a/internal/cli/dumps.go +++ b/internal/cli/dumps.go @@ -23,7 +23,7 @@ func dumpsListCmd() *cobra.Command { if err != nil { return err } - all, err := deps.Dumps.List() + all, err := deps.Dumps.List(c.Context()) if err != nil { return err } @@ -44,7 +44,7 @@ func dumpsInspectCmd() *cobra.Command { if err != nil { return err } - m, err := deps.Dumps.ReadMeta(args[0]) + m, err := deps.Dumps.ReadMeta(c.Context(), args[0]) if err != nil { return err } @@ -65,13 +65,13 @@ func dumpsPruneCmd() *cobra.Command { if err != nil { return err } - rep, err := deps.Dumps.PruneDryRun(dumps.RetentionPolicy{MaxAge: maxAge}) + rep, err := deps.Dumps.PruneDryRun(c.Context(), dumps.RetentionPolicy{MaxAge: maxAge}) if err != nil { return err } for _, m := range rep.Would { if apply { - if delErr := deps.Dumps.Delete(m.ID); delErr != nil { + if delErr := deps.Dumps.Delete(c.Context(), m.ID); delErr != nil { _, _ = fmt.Fprintf(c.OutOrStdout(), " ! failed to delete %s: %v\n", m.ID, delErr) continue } diff --git a/internal/cli/root.go b/internal/cli/root.go index 2763a58..4c38332 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -3,6 +3,7 @@ package cli import ( + "context" "errors" "fmt" "io" @@ -18,6 +19,7 @@ import ( "github.com/nixrajput/siphon/internal/jobs" "github.com/nixrajput/siphon/internal/profile" "github.com/nixrajput/siphon/internal/secrets" + "github.com/nixrajput/siphon/internal/storage" "github.com/nixrajput/siphon/internal/tui" ) @@ -72,18 +74,11 @@ func buildDeps() (app.Deps, error) { res := secrets.NewResolver(secrets.Env{}, secrets.Passthrough{}) ps := profile.New(cfg, res, config.Save) - dumpDir := cfg.Defaults.DumpDir - if dumpDir == "" { - home, homeErr := os.UserHomeDir() - if homeErr != nil { - return app.Deps{}, fmt.Errorf("resolve home dir for default dump_dir: %w", homeErr) - } - dumpDir = filepath.Join(home, ".local", "share", "siphon", "dumps") - } - cat, err := dumps.NewCatalog(dumpDir) + store, err := buildStore(cfg) if err != nil { return app.Deps{}, err } + cat := dumps.New(store) return app.Deps{ Profiles: ps, @@ -93,6 +88,34 @@ func buildDeps() (app.Deps, error) { }, nil } +// buildStore selects the dump-catalog storage backend from config. Type "s3" +// builds an S3-backed store; anything else (the default) uses the local +// filesystem rooted at Defaults.DumpDir (or the XDG share dir when unset). +func buildStore(cfg *config.Config) (storage.Store, error) { + if cfg.Storage.Type == "s3" { + st, err := storage.NewS3(context.Background(), storage.S3Options{ + Bucket: cfg.Storage.Bucket, + Prefix: cfg.Storage.Prefix, + Region: cfg.Storage.Region, + Endpoint: cfg.Storage.Endpoint, + }) + if err != nil { + return nil, fmt.Errorf("init s3 storage: %w", err) + } + return st, nil + } + + dumpDir := cfg.Defaults.DumpDir + if dumpDir == "" { + home, homeErr := os.UserHomeDir() + if homeErr != nil { + return nil, fmt.Errorf("resolve home dir for default dump_dir: %w", homeErr) + } + dumpDir = filepath.Join(home, ".local", "share", "siphon", "dumps") + } + return storage.NewLocal(dumpDir) +} + func newScheduleCmd() *cobra.Command { return &cobra.Command{ Use: "schedule", diff --git a/internal/config/config.go b/internal/config/config.go index 75cc72a..1f62397 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "os" + "strings" "gopkg.in/yaml.v3" ) @@ -12,6 +13,7 @@ import ( type Config struct { Version int `yaml:"version"` Defaults Defaults `yaml:"defaults"` + Storage StorageConfig `yaml:"storage"` Profiles map[string]ProfileConfig `yaml:"profiles"` Groups map[string]GroupConfig `yaml:"groups"` } @@ -23,6 +25,37 @@ type Defaults struct { SecretBackend string `yaml:"secret_backend"` } +// StorageConfig selects where the dump catalog physically lives. Type "local" +// (or empty, the default) uses Defaults.DumpDir on the local filesystem. Type +// "s3" stores dumps in an S3 or S3-compatible bucket. Credentials are NOT held +// here — the S3 SDK resolves them from the standard chain (env vars, shared +// config, instance role), so the config file stays free of secrets. +type StorageConfig struct { + Type string `yaml:"type"` // "local" | "s3" (default "local") + Bucket string `yaml:"bucket,omitempty"` // s3: target bucket + Prefix string `yaml:"prefix,omitempty"` // s3: optional key prefix within the bucket + Region string `yaml:"region,omitempty"` // s3: AWS region + Endpoint string `yaml:"endpoint,omitempty"` // s3: custom endpoint for S3-compatible services (MinIO, R2) +} + +// Validate checks the storage block for internal consistency. It is called by +// Load so a malformed storage config fails fast with a clear message rather +// than surfacing as an obscure runtime error on first backup. +func (s StorageConfig) Validate() error { + t := strings.TrimSpace(s.Type) + switch t { + case "", "local": + return nil + case "s3": + if strings.TrimSpace(s.Bucket) == "" { + return errors.New("storage.type is s3 but storage.bucket is empty") + } + return nil + default: + return fmt.Errorf("unknown storage.type %q (want \"local\" or \"s3\")", t) + } +} + // ProfileConfig is the unresolved on-disk form of a connection profile. // Name is NOT read from YAML — it is populated by Load() from the map key // in Config.Profiles so callers don't need to thread the name separately. @@ -80,6 +113,10 @@ func Load() (*Config, error) { cfg.Profiles[name] = p } + if err := cfg.Storage.Validate(); err != nil { + return nil, fmt.Errorf("config storage: %w", err) + } + return cfg, nil } diff --git a/internal/config/storage_test.go b/internal/config/storage_test.go new file mode 100644 index 0000000..3958818 --- /dev/null +++ b/internal/config/storage_test.go @@ -0,0 +1,25 @@ +package config + +import "testing" + +func TestStorageConfig_Validate(t *testing.T) { + cases := []struct { + name string + cfg StorageConfig + wantErr bool + }{ + {"empty defaults to local", StorageConfig{}, false}, + {"explicit local", StorageConfig{Type: "local"}, false}, + {"s3 with bucket", StorageConfig{Type: "s3", Bucket: "dumps"}, false}, + {"s3 without bucket", StorageConfig{Type: "s3"}, true}, + {"unknown type", StorageConfig{Type: "gcs"}, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := tc.cfg.Validate() + if (err != nil) != tc.wantErr { + t.Errorf("Validate() error = %v, wantErr %v", err, tc.wantErr) + } + }) + } +} diff --git a/internal/dumps/catalog.go b/internal/dumps/catalog.go index 3c88db3..bf1b754 100644 --- a/internal/dumps/catalog.go +++ b/internal/dumps/catalog.go @@ -1,32 +1,48 @@ package dumps import ( + "context" "encoding/json" "errors" - "os" - "path/filepath" + "io" "strings" "github.com/nixrajput/siphon/internal/errs" + "github.com/nixrajput/siphon/internal/storage" ) -// Catalog is a filesystem-backed dump store. Phase G replaces the -// filesystem path with a Storage abstraction; the API does not change. +// Catalog is a dump store addressed by dump ID. It holds a storage.Store +// substrate (local directory or object store) and maps each ID to two keys: +// ".dump" (the dump body, envelope-prefixed) and ".meta.json" (the +// sidecar metadata). Callers never see storage paths — only IDs. type Catalog struct { - root string + store storage.Store } -// NewCatalog creates the directory if missing and returns a Catalog rooted there. +const ( + dumpSuffix = ".dump" + metaSuffix = ".meta.json" +) + +// New returns a Catalog over the given storage backend. +func New(store storage.Store) *Catalog { + return &Catalog{store: store} +} + +// NewCatalog returns a Catalog backed by a local directory, creating it if +// missing. Retained for callers (and tests) that want the default local +// backend without constructing a storage.Store themselves. func NewCatalog(root string) (*Catalog, error) { - if err := os.MkdirAll(root, 0o700); err != nil { + st, err := storage.NewLocal(root) + if err != nil { return nil, err } - return &Catalog{root: root}, nil + return &Catalog{store: st}, nil } -// validID rejects IDs that could escape the catalog root when joined into a -// path. Dump IDs are ULIDs, but restore/inspect/verify take the ID from user -// input, so a crafted value like "../../etc/passwd" must not traverse out. +// validID rejects IDs that could escape the storage key space. Dump IDs are +// ULIDs, but restore/inspect/verify take the ID from user input, so a crafted +// value like "../../etc/passwd" must not traverse out. func validID(id string) error { if id == "" || id == "." || id == ".." || strings.ContainsAny(id, `/\`) || strings.Contains(id, "..") { @@ -35,73 +51,123 @@ func validID(id string) error { return nil } -// Path returns the dump file path for the given ID. filepath.Base defensively -// strips any path components so a stray separator can never escape root. -func (c *Catalog) Path(id string) string { return filepath.Join(c.root, filepath.Base(id)+".dump") } +func dumpKey(id string) string { return id + dumpSuffix } +func metaKey(id string) string { return id + metaSuffix } -// MetaPath returns the sidecar JSON path for the given ID. -func (c *Catalog) MetaPath(id string) string { - return filepath.Join(c.root, filepath.Base(id)+".meta.json") +// PutDump streams r (the envelope-prefixed dump body) to the store under the +// dump key for id. The store guarantees the object is published atomically, so a +// failed or cancelled write never leaves a partial dump addressable by id. +func (c *Catalog) PutDump(ctx context.Context, id string, r io.Reader) error { + if err := validID(id); err != nil { + return err + } + return c.store.Put(ctx, dumpKey(id), r) } -// Root returns the catalog's root directory. -func (c *Catalog) Root() string { return c.root } +// OpenDump opens the dump body for id as a one-shot forward stream. The caller +// must Close it. A missing dump maps to a CodeUser error. +func (c *Catalog) OpenDump(ctx context.Context, id string) (io.ReadCloser, error) { + if err := validID(id); err != nil { + return nil, err + } + rc, err := c.store.Get(ctx, dumpKey(id)) + if errors.Is(err, storage.ErrNotFound) { + return nil, &errs.Error{Op: "dumps.open", Code: errs.CodeUser, Cause: errs.ErrDumpCorrupt, Hint: "no dump body for " + id} + } + if err != nil { + return nil, &errs.Error{Op: "dumps.open", Code: errs.CodeSystem, Cause: err} + } + return rc, nil +} -// WriteMeta serializes m and writes it to .meta.json. -func (c *Catalog) WriteMeta(m *Meta) error { +// WriteMeta serializes m and writes it under the meta key. Write meta LAST when +// publishing a dump: the catalog enumerates by meta, so a dump body without its +// meta is an invisible (prunable) orphan, whereas a meta without its body would +// be a dangling catalog entry. +func (c *Catalog) WriteMeta(ctx context.Context, m *Meta) error { + if err := validID(m.ID); err != nil { + return err + } body, err := json.MarshalIndent(m, "", " ") if err != nil { return err } - return os.WriteFile(c.MetaPath(m.ID), body, 0o600) + return c.store.Put(ctx, metaKey(m.ID), strings.NewReader(string(body))) } -// ReadMeta loads .meta.json from disk. -func (c *Catalog) ReadMeta(id string) (*Meta, error) { +// ReadMeta loads and decodes the sidecar metadata for id. +func (c *Catalog) ReadMeta(ctx context.Context, id string) (*Meta, error) { if err := validID(id); err != nil { return nil, err } - body, err := os.ReadFile(c.MetaPath(id)) - if errors.Is(err, os.ErrNotExist) { + rc, err := c.store.Get(ctx, metaKey(id)) + if errors.Is(err, storage.ErrNotFound) { return nil, &errs.Error{Op: "dumps.read_meta", Code: errs.CodeUser, Cause: errs.ErrDumpCorrupt, Hint: "no metadata for " + id} } if err != nil { - return nil, err + return nil, &errs.Error{Op: "dumps.read_meta", Code: errs.CodeSystem, Cause: err} + } + defer func() { _ = rc.Close() }() + body, err := io.ReadAll(rc) + if err != nil { + return nil, &errs.Error{Op: "dumps.read_meta", Code: errs.CodeSystem, Cause: err} } m := &Meta{} - return m, json.Unmarshal(body, m) + if err := json.Unmarshal(body, m); err != nil { + return nil, &errs.Error{Op: "dumps.read_meta", Code: errs.CodeUser, Cause: errs.ErrDumpCorrupt, Hint: "metadata for " + id + " is corrupt"} + } + return m, nil } -// List returns metadata for every dump in the catalog, sorted newest first. -func (c *Catalog) List() ([]Meta, error) { - entries, err := os.ReadDir(c.root) +// List returns metadata for every dump in the catalog, sorted newest first. It +// enumerates meta keys, reading each; a corrupt entry is skipped rather than +// failing the whole listing. +func (c *Catalog) List(ctx context.Context) ([]Meta, error) { + keys, err := c.store.List(ctx) if err != nil { - return nil, err + return nil, &errs.Error{Op: "dumps.list", Code: errs.CodeSystem, Cause: err} } var out []Meta - for _, e := range entries { - if e.IsDir() || !strings.HasSuffix(e.Name(), ".meta.json") { + for _, k := range keys { + if !strings.HasSuffix(k, metaSuffix) { continue } - id := strings.TrimSuffix(e.Name(), ".meta.json") - m, err := c.ReadMeta(id) + id := strings.TrimSuffix(k, metaSuffix) + m, err := c.ReadMeta(ctx, id) if err != nil { - continue // skip corrupt entries; user can `dumps prune --orphans` later + // Skip only a corrupt/missing user entry (the user can prune it + // later). A system error — a backend failure or a cancelled context — + // must propagate, or List would return partial results as success and + // silently hide an outage or an aborted run. + var e *errs.Error + if errors.As(err, &e) && e.Code == errs.CodeUser { + continue + } + return nil, err } out = append(out, *m) } - // sort newest first sortByCreatedDesc(out) return out, nil } -// Delete removes both the dump file and its sidecar. -func (c *Catalog) Delete(id string) error { +// Delete removes both the dump body and its sidecar. Delete is idempotent in the +// store, so a missing object is not an error. +func (c *Catalog) Delete(ctx context.Context, id string) error { if err := validID(id); err != nil { return err } - _ = os.Remove(c.Path(id)) - return os.Remove(c.MetaPath(id)) + // Delete meta FIRST. The catalog enumerates by meta, so if the second delete + // fails the worst case is an invisible orphan body (prunable) rather than a + // catalog entry that lists a dump whose body is already gone. This mirrors + // the write path, which writes meta last for the same invariant. + if err := c.store.Delete(ctx, metaKey(id)); err != nil { + return &errs.Error{Op: "dumps.delete", Code: errs.CodeSystem, Cause: err} + } + if err := c.store.Delete(ctx, dumpKey(id)); err != nil { + return &errs.Error{Op: "dumps.delete", Code: errs.CodeSystem, Cause: err} + } + return nil } func sortByCreatedDesc(m []Meta) { diff --git a/internal/dumps/catalog_test.go b/internal/dumps/catalog_test.go index 6e41cfd..8c121b7 100644 --- a/internal/dumps/catalog_test.go +++ b/internal/dumps/catalog_test.go @@ -1,6 +1,7 @@ package dumps import ( + "context" "errors" "os" "path/filepath" @@ -25,7 +26,7 @@ func TestCatalog_WriteRead_Roundtrip(t *testing.T) { Checksum: "sha256:abc", SizeBytes: 1234, } - if err := c.WriteMeta(m); err != nil { + if err := c.WriteMeta(context.Background(), m); err != nil { t.Fatal(err) } @@ -33,7 +34,7 @@ func TestCatalog_WriteRead_Roundtrip(t *testing.T) { t.Fatalf("expected sidecar to exist: %v", statErr) } - got, err := c.ReadMeta(m.ID) + got, err := c.ReadMeta(context.Background(), m.ID) if err != nil { t.Fatal(err) } @@ -48,10 +49,10 @@ func TestCatalog_List_SortsNewestFirst(t *testing.T) { old := &Meta{ID: "01HOLD0000000000000000000", Created: time.Now().Add(-24 * time.Hour)} new_ := &Meta{ID: "01HNEW0000000000000000000", Created: time.Now()} - _ = c.WriteMeta(old) - _ = c.WriteMeta(new_) + _ = c.WriteMeta(context.Background(), old) + _ = c.WriteMeta(context.Background(), new_) - got, err := c.List() + got, err := c.List(context.Background()) if err != nil { t.Fatal(err) } @@ -66,10 +67,10 @@ func TestCatalog_PruneDryRun_AppliesMaxAge(t *testing.T) { old := &Meta{ID: "01HOLD0000000000000000000", Created: time.Now().Add(-48 * time.Hour)} new_ := &Meta{ID: "01HNEW0000000000000000000", Created: time.Now()} - _ = c.WriteMeta(old) - _ = c.WriteMeta(new_) + _ = c.WriteMeta(context.Background(), old) + _ = c.WriteMeta(context.Background(), new_) - rep, err := c.PruneDryRun(RetentionPolicy{MaxAge: 24 * time.Hour}) + rep, err := c.PruneDryRun(context.Background(), RetentionPolicy{MaxAge: 24 * time.Hour}) if err != nil { t.Fatal(err) } @@ -89,16 +90,16 @@ func TestCatalog_RejectsTraversalID(t *testing.T) { t.Fatal(err) } for _, bad := range []string{"../escape", "../../etc/passwd", "a/b", `a\b`, "", "..", "."} { - if _, err := c.ReadMeta(bad); err == nil { + if _, err := c.ReadMeta(context.Background(), bad); err == nil { t.Errorf("ReadMeta(%q) = nil error; want rejection", bad) } - if err := c.Delete(bad); err == nil { + if err := c.Delete(context.Background(), bad); err == nil { t.Errorf("Delete(%q) = nil error; want rejection", bad) } } // A clean ULID-like id passes validation, so ReadMeta proceeds and fails // only because the dump doesn't exist — surfaced as errs.ErrDumpCorrupt. - _, err = c.ReadMeta("01HXKZ000000000000000000") + _, err = c.ReadMeta(context.Background(), "01HXKZ000000000000000000") if err == nil { t.Fatal("ReadMeta(valid-but-missing id) = nil error; want a not-found error") } diff --git a/internal/dumps/chain.go b/internal/dumps/chain.go index 62c823a..36ea865 100644 --- a/internal/dumps/chain.go +++ b/internal/dumps/chain.go @@ -1,6 +1,7 @@ package dumps import ( + "context" "errors" "fmt" ) @@ -11,7 +12,7 @@ import ( // It walks ParentID backwards until it reaches a base (BaseID == ID, or a // legacy empty BaseID). Cycles and broken chains (a missing parent) are // reported as errors rather than looping forever or silently truncating. -func (c *Catalog) ResolveChain(targetID string) ([]Meta, error) { +func (c *Catalog) ResolveChain(ctx context.Context, targetID string) ([]Meta, error) { visited := map[string]bool{} var chain []Meta @@ -22,7 +23,7 @@ func (c *Catalog) ResolveChain(targetID string) ([]Meta, error) { } visited[cur] = true - m, err := c.ReadMeta(cur) + m, err := c.ReadMeta(ctx, cur) if err != nil { return nil, fmt.Errorf("chain broken at %s: %w", cur, err) } diff --git a/internal/dumps/chain_test.go b/internal/dumps/chain_test.go index 8ba60ad..ae6c493 100644 --- a/internal/dumps/chain_test.go +++ b/internal/dumps/chain_test.go @@ -1,6 +1,7 @@ package dumps import ( + "context" "testing" "time" ) @@ -8,9 +9,9 @@ import ( func TestResolveChain_SingleBase(t *testing.T) { c, _ := NewCatalog(t.TempDir()) base := &Meta{ID: "base", BaseID: "base", Created: time.Now()} - _ = c.WriteMeta(base) + _ = c.WriteMeta(context.Background(), base) - chain, err := c.ResolveChain("base") + chain, err := c.ResolveChain(context.Background(), "base") if err != nil { t.Fatal(err) } @@ -21,10 +22,10 @@ func TestResolveChain_SingleBase(t *testing.T) { func TestResolveChain_BaseAndOneIncremental(t *testing.T) { c, _ := NewCatalog(t.TempDir()) - _ = c.WriteMeta(&Meta{ID: "base", BaseID: "base", Created: time.Now()}) - _ = c.WriteMeta(&Meta{ID: "inc1", BaseID: "base", ParentID: "base", Created: time.Now()}) + _ = c.WriteMeta(context.Background(), &Meta{ID: "base", BaseID: "base", Created: time.Now()}) + _ = c.WriteMeta(context.Background(), &Meta{ID: "inc1", BaseID: "base", ParentID: "base", Created: time.Now()}) - chain, err := c.ResolveChain("inc1") + chain, err := c.ResolveChain(context.Background(), "inc1") if err != nil { t.Fatal(err) } @@ -35,11 +36,11 @@ func TestResolveChain_BaseAndOneIncremental(t *testing.T) { func TestResolveChain_MultiIncremental(t *testing.T) { c, _ := NewCatalog(t.TempDir()) - _ = c.WriteMeta(&Meta{ID: "base", BaseID: "base", Created: time.Now()}) - _ = c.WriteMeta(&Meta{ID: "inc1", BaseID: "base", ParentID: "base", Created: time.Now()}) - _ = c.WriteMeta(&Meta{ID: "inc2", BaseID: "base", ParentID: "inc1", Created: time.Now()}) + _ = c.WriteMeta(context.Background(), &Meta{ID: "base", BaseID: "base", Created: time.Now()}) + _ = c.WriteMeta(context.Background(), &Meta{ID: "inc1", BaseID: "base", ParentID: "base", Created: time.Now()}) + _ = c.WriteMeta(context.Background(), &Meta{ID: "inc2", BaseID: "base", ParentID: "inc1", Created: time.Now()}) - chain, err := c.ResolveChain("inc2") + chain, err := c.ResolveChain(context.Background(), "inc2") if err != nil { t.Fatal(err) } @@ -50,10 +51,10 @@ func TestResolveChain_MultiIncremental(t *testing.T) { func TestResolveChain_DetectsCycle(t *testing.T) { c, _ := NewCatalog(t.TempDir()) - _ = c.WriteMeta(&Meta{ID: "a", BaseID: "x", ParentID: "b"}) - _ = c.WriteMeta(&Meta{ID: "b", BaseID: "x", ParentID: "a"}) + _ = c.WriteMeta(context.Background(), &Meta{ID: "a", BaseID: "x", ParentID: "b"}) + _ = c.WriteMeta(context.Background(), &Meta{ID: "b", BaseID: "x", ParentID: "a"}) - if _, err := c.ResolveChain("a"); err == nil { + if _, err := c.ResolveChain(context.Background(), "a"); err == nil { t.Fatal("expected cycle error") } } @@ -61,9 +62,9 @@ func TestResolveChain_DetectsCycle(t *testing.T) { func TestResolveChain_BrokenChain_MissingParent(t *testing.T) { c, _ := NewCatalog(t.TempDir()) // inc1 claims a parent that was never written. - _ = c.WriteMeta(&Meta{ID: "inc1", BaseID: "base", ParentID: "base", Created: time.Now()}) + _ = c.WriteMeta(context.Background(), &Meta{ID: "inc1", BaseID: "base", ParentID: "base", Created: time.Now()}) - if _, err := c.ResolveChain("inc1"); err == nil { + if _, err := c.ResolveChain(context.Background(), "inc1"); err == nil { t.Fatal("expected broken-chain error (missing base)") } } diff --git a/internal/dumps/prune.go b/internal/dumps/prune.go index d53fb18..5260f74 100644 --- a/internal/dumps/prune.go +++ b/internal/dumps/prune.go @@ -1,6 +1,9 @@ package dumps -import "time" +import ( + "context" + "time" +) // RetentionPolicy is filled in by Phase G. Phase B ships only a dry-run prune // that lists what *would* be deleted given a max-age policy. @@ -16,8 +19,8 @@ type PruneReport struct { // PruneDryRun returns which dumps would be deleted under policy p without // actually deleting anything. -func (c *Catalog) PruneDryRun(p RetentionPolicy) (PruneReport, error) { - all, err := c.List() +func (c *Catalog) PruneDryRun(ctx context.Context, p RetentionPolicy) (PruneReport, error) { + all, err := c.List(ctx) if err != nil { return PruneReport{}, err } diff --git a/internal/storage/local.go b/internal/storage/local.go new file mode 100644 index 0000000..6d63832 --- /dev/null +++ b/internal/storage/local.go @@ -0,0 +1,198 @@ +package storage + +import ( + "context" + "errors" + "fmt" + "io" + "io/fs" + "os" + "path/filepath" + "strings" +) + +// localStore is a Store backed by a single local directory. Keys map directly to +// file names within root. Put is made atomic with the classic write-temp + +// rename dance, so a reader that fails mid-stream never leaves a partial object +// visible under its final name. +type localStore struct { + root string +} + +// NewLocal returns a Store backed by the directory root, creating it (0700) if +// absent. This preserves siphon's pre-Phase-G on-disk layout: keys are written +// verbatim as files under root, so an existing local catalog keeps working with +// no migration. +func NewLocal(root string) (Store, error) { + if err := os.MkdirAll(root, 0o700); err != nil { + return nil, fmt.Errorf("storage.local: mkdir %s: %w", root, err) + } + return &localStore{root: root}, nil +} + +// safeName rejects keys that could escape root once joined. Keys are +// siphon-internal (".dump"), but defense-in-depth here mirrors the catalog's +// own validID guard. +func (s *localStore) path(key string) (string, error) { + if key == "" || strings.ContainsAny(key, `/\`) || strings.Contains(key, "..") { + return "", fmt.Errorf("storage.local: invalid key %q", key) + } + return filepath.Join(s.root, key), nil +} + +func (s *localStore) Put(ctx context.Context, key string, r io.Reader) error { + p, err := s.path(key) + if err != nil { + return err + } + tmp, err := os.CreateTemp(s.root, ".tmp-"+key+"-*") + if err != nil { + return fmt.Errorf("storage.local: temp for %s: %w", key, err) + } + tmpName := tmp.Name() + // Copy with cancellation: a cancelled context aborts before the rename, so no + // partial object becomes visible under the final key. + if _, err := copyCtx(ctx, tmp, r); err != nil { + _ = tmp.Close() + _ = os.Remove(tmpName) + return fmt.Errorf("storage.local: write %s: %w", key, err) + } + // fsync the data before publishing: the Store contract promises a durable + // write, and this is a backup tool — a dump that survives `Put` but is lost + // to a power cut defeats the purpose. Sync the file, then (below) the parent + // dir so the rename itself is durable. + if err := tmp.Sync(); err != nil { + _ = tmp.Close() + _ = os.Remove(tmpName) + return fmt.Errorf("storage.local: sync %s: %w", key, err) + } + if err := tmp.Close(); err != nil { + _ = os.Remove(tmpName) + return fmt.Errorf("storage.local: flush %s: %w", key, err) + } + if err := atomicRename(tmpName, p); err != nil { + _ = os.Remove(tmpName) + return fmt.Errorf("storage.local: publish %s: %w", key, err) + } + if err := syncDir(s.root); err != nil { + return fmt.Errorf("storage.local: sync dir for %s: %w", key, err) + } + return nil +} + +// atomicRename renames src→dst, replacing dst if it exists. os.Rename already +// replaces an existing destination on Unix; on Windows/some filesystems it can +// fail specifically because dst exists, so in THAT case only we remove dst and +// retry. The guard is deliberately narrow: removing dst on an unrelated error +// (transient I/O, permissions, device busy) would destroy the last good copy of +// the dump and still fail — worse than the rename failure itself. +func atomicRename(src, dst string) error { + err := os.Rename(src, dst) + if err == nil { + return nil + } + if !errors.Is(err, fs.ErrExist) { + return err // not a "destination exists" failure — do NOT touch dst + } + if removeErr := os.Remove(dst); removeErr != nil && !errors.Is(removeErr, fs.ErrNotExist) { + return err // could not clear dst; surface the original rename error + } + return os.Rename(src, dst) +} + +// syncDir fsyncs a directory so a rename into it is durable across a crash. A +// directory that cannot be opened or synced (some platforms/filesystems reject +// directory fsync) is tolerated — the data file itself was already synced. +func syncDir(dir string) error { + d, err := os.Open(dir) + if err != nil { + return nil //nolint:nilerr // best-effort dir sync; data fsync already done + } + defer func() { _ = d.Close() }() + if err := d.Sync(); err != nil { + return nil //nolint:nilerr // some filesystems reject dir fsync; tolerate it + } + return nil +} + +func (s *localStore) Get(_ context.Context, key string) (io.ReadCloser, error) { + p, err := s.path(key) + if err != nil { + return nil, err + } + f, err := os.Open(p) + if os.IsNotExist(err) { + return nil, fmt.Errorf("storage.local: %s: %w", key, ErrNotFound) + } + if err != nil { + return nil, fmt.Errorf("storage.local: open %s: %w", key, err) + } + return f, nil +} + +func (s *localStore) Delete(_ context.Context, key string) error { + p, err := s.path(key) + if err != nil { + return err + } + if err := os.Remove(p); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("storage.local: delete %s: %w", key, err) + } + return nil // idempotent: a missing key is not an error +} + +func (s *localStore) List(_ context.Context) ([]string, error) { + entries, err := os.ReadDir(s.root) + if err != nil { + return nil, fmt.Errorf("storage.local: list %s: %w", s.root, err) + } + var keys []string + for _, e := range entries { + if e.IsDir() || strings.HasPrefix(e.Name(), ".tmp-") { + continue // skip subdirs and in-flight temp files + } + keys = append(keys, e.Name()) + } + return keys, nil +} + +func (s *localStore) Stat(_ context.Context, key string) (int64, bool, error) { + p, err := s.path(key) + if err != nil { + return 0, false, err + } + fi, err := os.Stat(p) + if os.IsNotExist(err) { + return 0, false, nil + } + if err != nil { + return 0, false, fmt.Errorf("storage.local: stat %s: %w", key, err) + } + return fi.Size(), true, nil +} + +// copyCtx copies r→w like io.Copy but aborts when ctx is cancelled. It checks +// the context between chunks so a long upload/download stops promptly. +func copyCtx(ctx context.Context, w io.Writer, r io.Reader) (int64, error) { + buf := make([]byte, 32*1024) + var total int64 + for { + if err := ctx.Err(); err != nil { + return total, err + } + n, rerr := r.Read(buf) + if n > 0 { + wn, werr := w.Write(buf[:n]) + total += int64(wn) + if werr != nil { + return total, werr + } + } + if rerr == io.EOF { + return total, nil + } + if rerr != nil { + return total, rerr + } + } +} diff --git a/internal/storage/local_test.go b/internal/storage/local_test.go new file mode 100644 index 0000000..df3bf24 --- /dev/null +++ b/internal/storage/local_test.go @@ -0,0 +1,113 @@ +package storage + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestLocalStore_Contract(t *testing.T) { + RunStoreSuite(t, func(t *testing.T) Store { + s, err := NewLocal(t.TempDir()) + if err != nil { + t.Fatalf("NewLocal: %v", err) + } + return s + }) +} + +// TestLocalStore_AtomicPublish proves Put never leaves a partial object under +// the final key: a reader that errors mid-stream must abort before the rename, +// so the key does not resolve. +func TestLocalStore_AtomicPublish(t *testing.T) { + dir := t.TempDir() + s, err := NewLocal(dir) + if err != nil { + t.Fatalf("NewLocal: %v", err) + } + err = s.Put(context.Background(), "x.dump", &failingReader{after: 4}) + if err == nil { + t.Fatal("Put with failing reader returned nil, want error") + } + if _, exists, _ := s.Stat(context.Background(), "x.dump"); exists { + t.Error("partial object visible under final key after failed Put") + } + // And no leftover temp file should linger. + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("ReadDir: %v", err) + } + if len(entries) != 0 { + var names []string + for _, e := range entries { + names = append(names, e.Name()) + } + t.Errorf("dir not clean after failed Put: %v", names) + } +} + +// TestLocalStore_KeyLayout pins the on-disk layout: a key maps verbatim to a +// file of that name under root, so pre-Phase-G local catalogs keep working. +func TestLocalStore_KeyLayout(t *testing.T) { + dir := t.TempDir() + s, err := NewLocal(dir) + if err != nil { + t.Fatalf("NewLocal: %v", err) + } + if err := s.Put(context.Background(), "01H.dump", strings.NewReader("body")); err != nil { + t.Fatalf("Put: %v", err) + } + if _, err := os.Stat(filepath.Join(dir, "01H.dump")); err != nil { + t.Errorf("expected file %s on disk: %v", filepath.Join(dir, "01H.dump"), err) + } +} + +// TestAtomicRename_PreservesDstOnUnrelatedError proves the narrow guard: when a +// rename fails for a reason OTHER than "destination exists" (here, a missing +// source → ErrNotExist), atomicRename must NOT delete the existing destination. +// The earlier broad "remove dst on any error" would have destroyed the last +// good copy of a dump on a transient glitch. +func TestAtomicRename_PreservesDstOnUnrelatedError(t *testing.T) { + dir := t.TempDir() + dst := filepath.Join(dir, "good.dump") + if err := os.WriteFile(dst, []byte("the only good copy"), 0o600); err != nil { + t.Fatalf("seed dst: %v", err) + } + missingSrc := filepath.Join(dir, "does-not-exist.tmp") + + if err := atomicRename(missingSrc, dst); err == nil { + t.Fatal("atomicRename with missing source returned nil, want error") + } + // The existing destination must still be intact. + got, err := os.ReadFile(dst) + if err != nil { + t.Fatalf("dst was destroyed by a failed rename: %v", err) + } + if string(got) != "the only good copy" { + t.Errorf("dst content = %q, want it untouched", got) + } +} + +// failingReader returns `after` bytes then a non-EOF error, simulating a body +// source that dies mid-stream. +type failingReader struct { + after int + n int +} + +func (r *failingReader) Read(p []byte) (int, error) { + if r.n >= r.after { + return 0, errReadFailed + } + p[0] = 'x' + r.n++ + return 1, nil +} + +var errReadFailed = osErr("simulated read failure") + +type osErr string + +func (e osErr) Error() string { return string(e) } diff --git a/internal/storage/s3.go b/internal/storage/s3.go new file mode 100644 index 0000000..449403d --- /dev/null +++ b/internal/storage/s3.go @@ -0,0 +1,171 @@ +package storage + +import ( + "context" + "errors" + "fmt" + "io" + "strings" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/feature/s3/transfermanager" + "github.com/aws/aws-sdk-go-v2/service/s3" + s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/aws/smithy-go" +) + +// S3Options configures an S3 (or S3-compatible) Store. Credentials are NOT here: +// the SDK resolves them from the standard chain (env vars, shared config, +// instance/role), keeping the siphon config file free of secrets. +type S3Options struct { + Bucket string + Prefix string // optional key prefix within the bucket + Region string + Endpoint string // optional custom endpoint for S3-compatible services (MinIO, R2) +} + +// s3Store is a Store backed by an S3 bucket. Put streams via the multipart +// upload manager (atomic on completion — the key does not resolve until the +// upload finishes), so a failed/cancelled upload leaves no partial object. +type s3Store struct { + client *s3.Client + uploader *transfermanager.Client + bucket string + prefix string +} + +// NewS3 builds an S3-backed Store. It loads AWS config (region, credentials) +// from the default chain and applies the optional custom endpoint for +// S3-compatible services. The bucket is assumed to exist. +func NewS3(ctx context.Context, opt S3Options) (Store, error) { + if opt.Bucket == "" { + return nil, errors.New("storage.s3: bucket is required") + } + var loadOpts []func(*awsconfig.LoadOptions) error + if opt.Region != "" { + loadOpts = append(loadOpts, awsconfig.WithRegion(opt.Region)) + } + cfg, err := awsconfig.LoadDefaultConfig(ctx, loadOpts...) + if err != nil { + return nil, fmt.Errorf("storage.s3: load aws config: %w", err) + } + client := s3.NewFromConfig(cfg, func(o *s3.Options) { + if opt.Endpoint != "" { + o.BaseEndpoint = aws.String(opt.Endpoint) + o.UsePathStyle = true // S3-compatible services (MinIO) need path-style + } + }) + return &s3Store{ + client: client, + uploader: transfermanager.New(client), + bucket: opt.Bucket, + prefix: strings.TrimSuffix(opt.Prefix, "/"), + }, nil +} + +func (s *s3Store) objectKey(key string) string { + if s.prefix == "" { + return key + } + return s.prefix + "/" + key +} + +func (s *s3Store) Put(ctx context.Context, key string, r io.Reader) error { + _, err := s.uploader.UploadObject(ctx, &transfermanager.UploadObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(s.objectKey(key)), + Body: r, + }) + if err != nil { + return fmt.Errorf("storage.s3: put %s: %w", key, err) + } + return nil +} + +func (s *s3Store) Get(ctx context.Context, key string) (io.ReadCloser, error) { + out, err := s.client.GetObject(ctx, &s3.GetObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(s.objectKey(key)), + }) + if err != nil { + if isS3NotFound(err) { + return nil, fmt.Errorf("storage.s3: %s: %w", key, ErrNotFound) + } + return nil, fmt.Errorf("storage.s3: get %s: %w", key, err) + } + return out.Body, nil +} + +func (s *s3Store) Delete(ctx context.Context, key string) error { + _, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(s.objectKey(key)), + }) + if err != nil && !isS3NotFound(err) { + return fmt.Errorf("storage.s3: delete %s: %w", key, err) + } + return nil // idempotent +} + +func (s *s3Store) List(ctx context.Context) ([]string, error) { + var keys []string + prefix := "" + if s.prefix != "" { + prefix = s.prefix + "/" + } + p := s3.NewListObjectsV2Paginator(s.client, &s3.ListObjectsV2Input{ + Bucket: aws.String(s.bucket), + Prefix: aws.String(prefix), + }) + for p.HasMorePages() { + page, err := p.NextPage(ctx) + if err != nil { + return nil, fmt.Errorf("storage.s3: list: %w", err) + } + for _, obj := range page.Contents { + k := aws.ToString(obj.Key) + keys = append(keys, strings.TrimPrefix(k, prefix)) + } + } + return keys, nil +} + +func (s *s3Store) Stat(ctx context.Context, key string) (int64, bool, error) { + out, err := s.client.HeadObject(ctx, &s3.HeadObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(s.objectKey(key)), + }) + if err != nil { + if isS3NotFound(err) { + return 0, false, nil + } + return 0, false, fmt.Errorf("storage.s3: stat %s: %w", key, err) + } + return aws.ToInt64(out.ContentLength), true, nil +} + +// isS3NotFound reports whether err is S3's "no such key / not found". GetObject +// returns *types.NoSuchKey; HeadObject returns a generic *types.NotFound (it has +// no typed NoSuchKey). Both are matched here so missing-object handling is +// uniform across operations. +func isS3NotFound(err error) bool { + var nsk *s3types.NoSuchKey + if errors.As(err, &nsk) { + return true + } + var nf *s3types.NotFound + if errors.As(err, &nf) { + return true + } + // S3-compatible services sometimes surface a bare API error code instead of + // the typed shape; match the canonical codes as a fallback. + var apiErr smithy.APIError + if errors.As(err, &apiErr) { + switch apiErr.ErrorCode() { + case "NoSuchKey", "NotFound", "404": + return true + } + } + return false +} diff --git a/internal/storage/s3_integration_test.go b/internal/storage/s3_integration_test.go new file mode 100644 index 0000000..9c050d1 --- /dev/null +++ b/internal/storage/s3_integration_test.go @@ -0,0 +1,80 @@ +//go:build integration + +package storage + +import ( + "context" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" + tcminio "github.com/testcontainers/testcontainers-go/modules/minio" +) + +// TestS3Store_Contract runs the full Store contract suite against a real MinIO +// (S3-compatible) container, so the streaming multipart Put, GetObject Get, +// ListObjectsV2, HeadObject Stat, and not-found mapping all execute against a +// live object store — not just compile. Each subtest gets a fresh uniquely +// prefixed view of one shared bucket, so the suite's "empty store" assumption +// holds without standing up a container per subtest. +func TestS3Store_Contract(t *testing.T) { + ctx := context.Background() + + mc, err := tcminio.Run(ctx, "minio/minio:RELEASE.2024-01-16T16-07-38Z") + if err != nil { + t.Fatalf("start minio: %v", err) + } + t.Cleanup(func() { _ = mc.Terminate(ctx) }) + + endpoint, err := mc.ConnectionString(ctx) + if err != nil { + t.Fatalf("minio endpoint: %v", err) + } + endpointURL := "http://" + endpoint + + // MinIO credentials come from the container; feed them to the SDK via env so + // NewS3's default credential chain picks them up. + t.Setenv("AWS_ACCESS_KEY_ID", mc.Username) + t.Setenv("AWS_SECRET_ACCESS_KEY", mc.Password) + t.Setenv("AWS_REGION", "us-east-1") + + const bucket = "siphon-test" + createBucket(t, ctx, endpointURL, mc.Username, mc.Password, bucket) + + prefixN := 0 + RunStoreSuite(t, func(t *testing.T) Store { + // Unique prefix per subtest → a fresh empty view of the shared bucket. + prefixN++ + st, err := NewS3(ctx, S3Options{ + Bucket: bucket, + Prefix: "sub" + string(rune('a'+prefixN)), + Region: "us-east-1", + Endpoint: endpointURL, + }) + if err != nil { + t.Fatalf("NewS3: %v", err) + } + return st + }) +} + +// createBucket provisions the test bucket on the fresh MinIO instance. +func createBucket(t *testing.T, ctx context.Context, endpoint, user, pass, bucket string) { + t.Helper() + cfg, err := awsconfig.LoadDefaultConfig(ctx, + awsconfig.WithRegion("us-east-1"), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(user, pass, "")), + ) + if err != nil { + t.Fatalf("aws config: %v", err) + } + client := s3.NewFromConfig(cfg, func(o *s3.Options) { + o.BaseEndpoint = aws.String(endpoint) + o.UsePathStyle = true + }) + if _, err := client.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String(bucket)}); err != nil { + t.Fatalf("create bucket: %v", err) + } +} diff --git a/internal/storage/store.go b/internal/storage/store.go new file mode 100644 index 0000000..f4b7c88 --- /dev/null +++ b/internal/storage/store.go @@ -0,0 +1,58 @@ +// Package storage abstracts where dump objects and their sidecar metadata +// physically live. The dump catalog (internal/dumps) holds a Store and addresses +// objects by opaque keys (e.g. ".dump", ".meta.json") rather than +// filesystem paths, so the same catalog logic works over a local directory or an +// object store (S3 and S3-compatible services such as MinIO, R2). +// +// This package is a stdlib-only leaf: it defines the Store interface plus a +// shared not-found sentinel. Concrete backends live in sibling files (local.go, +// s3.go) and may import third-party SDKs; the interface itself does not. +package storage + +import ( + "context" + "errors" + "io" +) + +// ErrNotFound is returned by Get when the requested key is absent. (Delete and +// Stat treat a missing key as a non-error — see their contracts below.) +// Backends MUST wrap their native "no such object" error with this sentinel +// (via fmt.Errorf("...: %w", ErrNotFound) or by returning it directly) so the +// catalog and app layers can distinguish "you asked for something that isn't +// here" (a user error) from a transient transport failure (a system error). +var ErrNotFound = errors.New("storage: object not found") + +// Store is the durable key→bytes substrate behind the dump catalog. +// +// Keys are opaque, caller-chosen strings. A backend may map a key onto a file +// name or an object key, but callers never depend on that mapping. +// +// All methods take a context: object I/O is network I/O for remote backends and +// must be cancellable. A cancelled context aborts the operation. +type Store interface { + // Put writes the full contents of r under key, durably and atomically: the + // key either resolves to the complete object or does not resolve at all — a + // reader that fails mid-stream, or a cancelled context, must not leave a + // partial object visible under key. Overwriting an existing key is allowed + // and replaces it. Put reads r to EOF. + Put(ctx context.Context, key string, r io.Reader) error + + // Get opens key for reading. The returned ReadCloser is a one-shot forward + // stream — callers must not assume it is seekable — and must be closed. A + // missing key returns an error wrapping ErrNotFound. + Get(ctx context.Context, key string) (io.ReadCloser, error) + + // Delete removes key. Deleting a key that does not exist is NOT an error + // (delete is idempotent), so callers can prune without racing existence. + Delete(ctx context.Context, key string) error + + // List returns every key currently present in the store, in no guaranteed + // order. + List(ctx context.Context) ([]string, error) + + // Stat reports the size in bytes and existence of key. A missing key returns + // (0, false, nil) — absence is not an error for Stat. A transport failure + // returns a non-nil error. + Stat(ctx context.Context, key string) (size int64, exists bool, err error) +} diff --git a/internal/storage/suite.go b/internal/storage/suite.go new file mode 100644 index 0000000..f1aa5be --- /dev/null +++ b/internal/storage/suite.go @@ -0,0 +1,149 @@ +package storage + +import ( + "bytes" + "context" + "errors" + "io" + "sort" + "testing" +) + +// RunStoreSuite exercises the Store contract against a backend. Every Store +// implementation runs this same table, so correctness is a property of +// implementing the interface — not a per-backend afterthought. newStore returns +// a fresh, empty Store for each subtest (e.g. a t.TempDir-rooted local store, or +// a uniquely-prefixed bucket view). +// +// It lives in a non-test file so both the local unit test and the S3 integration +// test (separate build tags) can call it. +func RunStoreSuite(t *testing.T, newStore func(t *testing.T) Store) { + t.Helper() + ctx := context.Background() + + t.Run("put then get round-trips bytes", func(t *testing.T) { + s := newStore(t) + want := []byte("envelope+body payload\x00with NUL and \xff bytes") + if err := s.Put(ctx, "a.dump", bytes.NewReader(want)); err != nil { + t.Fatalf("Put: %v", err) + } + rc, err := s.Get(ctx, "a.dump") + if err != nil { + t.Fatalf("Get: %v", err) + } + defer func() { _ = rc.Close() }() + got, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !bytes.Equal(got, want) { + t.Errorf("round-trip mismatch: got %q want %q", got, want) + } + }) + + t.Run("get missing key wraps ErrNotFound", func(t *testing.T) { + s := newStore(t) + _, err := s.Get(ctx, "nope.dump") + if !errors.Is(err, ErrNotFound) { + t.Fatalf("Get missing: err = %v, want wrapping ErrNotFound", err) + } + }) + + t.Run("stat reports size and existence", func(t *testing.T) { + s := newStore(t) + body := []byte("0123456789") + if err := s.Put(ctx, "sz.dump", bytes.NewReader(body)); err != nil { + t.Fatalf("Put: %v", err) + } + size, exists, err := s.Stat(ctx, "sz.dump") + if err != nil { + t.Fatalf("Stat: %v", err) + } + if !exists || size != int64(len(body)) { + t.Errorf("Stat = (%d, %v), want (%d, true)", size, exists, len(body)) + } + // Missing key: (0, false, nil) — absence is not an error for Stat. + _, exists, err = s.Stat(ctx, "missing.dump") + if err != nil || exists { + t.Errorf("Stat missing = (exists %v, err %v), want (false, nil)", exists, err) + } + }) + + t.Run("delete is idempotent", func(t *testing.T) { + s := newStore(t) + if err := s.Put(ctx, "d.dump", bytes.NewReader([]byte("x"))); err != nil { + t.Fatalf("Put: %v", err) + } + if err := s.Delete(ctx, "d.dump"); err != nil { + t.Fatalf("Delete: %v", err) + } + if _, exists, _ := s.Stat(ctx, "d.dump"); exists { + t.Error("key still present after Delete") + } + // Deleting an absent key is not an error. + if err := s.Delete(ctx, "d.dump"); err != nil { + t.Errorf("Delete of absent key: %v, want nil", err) + } + }) + + t.Run("list returns present keys", func(t *testing.T) { + s := newStore(t) + for _, k := range []string{"one.dump", "one.meta.json", "two.dump"} { + if err := s.Put(ctx, k, bytes.NewReader([]byte("x"))); err != nil { + t.Fatalf("Put %s: %v", k, err) + } + } + keys, err := s.List(ctx) + if err != nil { + t.Fatalf("List: %v", err) + } + sort.Strings(keys) + want := []string{"one.dump", "one.meta.json", "two.dump"} + if len(keys) != len(want) { + t.Fatalf("List = %v, want %v", keys, want) + } + for i := range want { + if keys[i] != want[i] { + t.Errorf("List[%d] = %q, want %q", i, keys[i], want[i]) + } + } + }) + + t.Run("put overwrites existing key", func(t *testing.T) { + s := newStore(t) + if err := s.Put(ctx, "o.dump", bytes.NewReader([]byte("first"))); err != nil { + t.Fatalf("Put 1: %v", err) + } + if err := s.Put(ctx, "o.dump", bytes.NewReader([]byte("second"))); err != nil { + t.Fatalf("Put 2: %v", err) + } + rc, err := s.Get(ctx, "o.dump") + if err != nil { + t.Fatalf("Get: %v", err) + } + defer func() { _ = rc.Close() }() + got, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if string(got) != "second" { + t.Errorf("after overwrite got %q, want %q", got, "second") + } + }) + + t.Run("cancelled context aborts put without publishing", func(t *testing.T) { + s := newStore(t) + cctx, cancel := context.WithCancel(ctx) + cancel() // already cancelled + err := s.Put(cctx, "cancel.dump", bytes.NewReader([]byte("data"))) + if err == nil { + t.Fatal("Put with cancelled context returned nil, want error") + } + // The key must NOT be visible — a cancelled Put leaves no partial object. + if _, exists, statErr := s.Stat(ctx, "cancel.dump"); statErr != nil { + t.Fatalf("Stat after cancelled Put: %v", statErr) + } else if exists { + t.Error("cancelled Put left a partial object visible") + } + }) +} diff --git a/internal/tui/dashboard_test.go b/internal/tui/dashboard_test.go index b25f9aa..38f8030 100644 --- a/internal/tui/dashboard_test.go +++ b/internal/tui/dashboard_test.go @@ -1,6 +1,7 @@ package tui import ( + "context" "errors" "strings" "testing" @@ -134,7 +135,7 @@ func TestNewRestore_AdaptiveDumpIDField(t *testing.T) { Driver: "postgres", Created: time.Now(), } - if err := d.Dumps.WriteMeta(meta); err != nil { + if err := d.Dumps.WriteMeta(context.Background(), meta); err != nil { t.Fatalf("WriteMeta: %v", err) } form, res := modals.NewRestore(d, "", "") diff --git a/internal/tui/modals/restore.go b/internal/tui/modals/restore.go index ebb67a4..b070cbe 100644 --- a/internal/tui/modals/restore.go +++ b/internal/tui/modals/restore.go @@ -1,6 +1,9 @@ package modals import ( + "context" + "time" + "github.com/charmbracelet/huh" "github.com/nixrajput/siphon/internal/app" @@ -25,7 +28,11 @@ func NewRestore(d app.Deps, defaultProfile, defaultDump string) (*huh.Form, *Res res := &RestoreResult{Profile: defaultProfile, DumpID: defaultDump} var dumpIDField huh.Field - metas, err := d.Dumps.List() + // Bound the catalog list: with S3 storage this is a network call, and the + // TUI form must not hang the UI on a slow/unreachable backend. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + metas, err := d.Dumps.List(ctx) if err == nil && len(metas) > 0 { opts := make([]huh.Option[string], len(metas)) for i, m := range metas { diff --git a/internal/tui/panels/dumps.go b/internal/tui/panels/dumps.go index ddd1beb..a0155f5 100644 --- a/internal/tui/panels/dumps.go +++ b/internal/tui/panels/dumps.go @@ -1,6 +1,7 @@ package panels import ( + "context" "fmt" "time" @@ -81,7 +82,11 @@ func (p Dumps) SelectedID() string { // Reload rebuilds the rows from the catalog. func (p *Dumps) Reload() { - all, err := p.deps.Dumps.List() + // Bound the catalog list: with S3 storage this is a network call, and the + // dashboard reload must not hang on a slow/unreachable backend. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + all, err := p.deps.Dumps.List(ctx) if err != nil { return }