diff --git a/CHANGELOG.md b/CHANGELOG.md index 34f5130..f462a08 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,3 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `mysql` and `mariadb` drivers as thin wrappers that inject the fork-specific binary names and declare capabilities honestly (`Parallel: false` — the dump tools are single-threaded). Registered via side-effect import in `internal/app/drivers.go`. - Integration suites for both engines run on the Phase D `RunDriverSuite` harness via testcontainers (`mysql:8.0`, `mariadb:11`). - CI installs the Postgres/MySQL/MariaDB client tools and runs the full integration suite on the Ubuntu runner (where Docker is available); unit tests continue to run on Linux/macOS/Windows. +- Phase F advanced-transfer machinery (some CLI paths gated pending follow-up wiring — see the `docs/INCREMENTAL.md`, `docs/CROSS_ENGINE.md`, `docs/CDC.md` Status sections): + - **Working today:** every dump is prefixed with a 4 KB `SIPH` JSON envelope (type, base/parent IDs, WAL/binlog positions, checksum); `restore` resolves the base→incremental chain and applies it in order, with `--up-to ` to stop early (cycle + broken-chain detection). Sync now streams through a bounded `jobs.Stream` (default 64×1 MB) instead of `io.Pipe`, exposing a `FillPercent` backpressure metric and propagating backup failures to the restore side via `CloseErr` (no truncated dump committed as clean). + - **Machinery in place, CLI gated (follow-up):** incremental backup capture (Postgres WAL replication slot + LSN; MySQL/MariaDB binlog position) — `backup --incremental` returns a clear "not yet wired" error; cross-engine type-mapping (canonical schema + JSONL emit/consume with per-engine identifier quoting and placeholders) — `sync --cross-engine` is capability-gated off until typed schema introspection exists; CDC continuous mode (state-file persistence + resume + capability gating) ships as a polling scaffold, not real logical-replication streaming. + - Known follow-up gates before incremental backup ships to users: Postgres orphan-replication-slot cleanup, validation of the `pg_receivewal`/`mysqlbinlog` streaming invocations against live servers (compile-checked only here). diff --git a/README.md b/README.md index 63eda1b..fdd2840 100644 --- a/README.md +++ b/README.md @@ -47,16 +47,16 @@ A single binary that turns the painful, error-prone sprawl of `pg_dump` → `pg_ ## Project status -| Phase | What it delivers | Status | -| --------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------- | -| **A** — Skeleton | Go module, Cobra CLI, TUI placeholder, `Driver` interface + registry, `errs`/`config`/`secrets`/`profile` packages, golangci-lint + depguard, cross-platform CI | ✅ Complete | -| **B** — Postgres walking skeleton | `backup`, `restore`, `sync`, `verify`, `inspect`, `dumps`, `config`, `profile` working end-to-end against PostgreSQL | ✅ Complete | -| **C** — TUI dashboard | Multi-panel Bubble Tea dashboard (profiles · dumps · jobs) with live job progress, backup/restore modal forms, and snapshot tests | ✅ Complete | -| **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 | Incremental backups, bounded-buffer streaming, cross-engine sync, CDC | ⏳ Planned | -| **G** — Ops features | Cloud storage, secret backends, profile groups + 2FA, team mode, audit log, retention, telemetry | ⏳ Planned | -| **H** — Distribution | GoReleaser, Homebrew tap, Scoop bucket, install script, docs site | ⏳ Planned | +| Phase | What it delivers | Status | +| --------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------- | +| **A** — Skeleton | Go module, Cobra CLI, TUI placeholder, `Driver` interface + registry, `errs`/`config`/`secrets`/`profile` packages, golangci-lint + depguard, cross-platform CI | ✅ Complete | +| **B** — Postgres walking skeleton | `backup`, `restore`, `sync`, `verify`, `inspect`, `dumps`, `config`, `profile` working end-to-end against PostgreSQL | ✅ Complete | +| **C** — TUI dashboard | Multi-panel Bubble Tea dashboard (profiles · dumps · jobs) with live job progress, backup/restore modal forms, and snapshot tests | ✅ Complete | +| **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 | Bounded-buffer streaming sync + chain-walking incremental restore work today; the dump envelope, incremental WAL/binlog capture, cross-engine type-mapping, and CDC state machinery are in place but their CLI paths (`backup --incremental`, `sync --cross-engine`, CDC) are gated pending follow-up wiring — see [docs/INCREMENTAL.md](docs/INCREMENTAL.md), [docs/CROSS_ENGINE.md](docs/CROSS_ENGINE.md), [docs/CDC.md](docs/CDC.md) | 🟡 Partial | +| **G** — Ops features | Cloud storage, secret backends, profile groups + 2FA, team mode, audit log, retention, telemetry | ⏳ Planned | +| **H** — Distribution | GoReleaser, Homebrew tap, Scoop bucket, install script, docs site | ⏳ Planned | ## Requirements @@ -66,12 +66,12 @@ A single binary that turns the painful, error-prone sprawl of `pg_dump` → `pg_ - **MySQL** profiles need `mysqldump`, `mysql`. - **MariaDB** profiles need `mariadb-dump`, `mariadb` (the renamed binaries shipped by MariaDB 10.5+; older installs that only ship `mysqldump`/`mysql` are not yet supported). - | Platform | PostgreSQL | MySQL / MariaDB | - | ------------- | --------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------- | - | macOS | `brew install postgresql@16` | `brew install mysql-client` / `brew install mariadb` | - | Debian/Ubuntu | `sudo apt install postgresql-client` | `sudo apt install mysql-client mariadb-client` | - | Fedora/RHEL | `sudo dnf install postgresql` | `sudo dnf install mysql mariadb` | - | Windows | Install the [EDB PostgreSQL](https://www.postgresql.org/download/windows/) package and add its `bin/` to `PATH` | Install the MySQL / MariaDB client and add to `PATH` | + | Platform | PostgreSQL | MySQL / MariaDB | + | ------------- | --------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------- | + | macOS | `brew install postgresql@16` | `brew install mysql-client` / `brew install mariadb` | + | Debian/Ubuntu | `sudo apt install postgresql-client` | `sudo apt install mysql-client mariadb-client` | + | Fedora/RHEL | `sudo dnf install postgresql` | `sudo dnf install mysql mariadb` | + | Windows | Install the [EDB PostgreSQL](https://www.postgresql.org/download/windows/) package and add its `bin/` to `PATH` | Install the MySQL / MariaDB client and add to `PATH` | - **Docker** _(optional)_ — only needed to run the integration test suite (`make test-integration`). @@ -222,6 +222,8 @@ Contributions are welcome. Please read [CONTRIBUTING.md](CONTRIBUTING.md), keep Adding a new database engine? See [docs/DRIVERS.md](docs/DRIVERS.md) for the driver contributor guide. +Concept docs (honest as-built status — several paths are documented follow-ups): [docs/INCREMENTAL.md](docs/INCREMENTAL.md) (incremental restore works; the backup path is a follow-up), [docs/CROSS_ENGINE.md](docs/CROSS_ENGINE.md) (type-map matrix; `--cross-engine` is gated off pending typed introspection), and [docs/CDC.md](docs/CDC.md) (scaffold only; not yet runnable). + ## License [MIT](LICENSE) © [Nikhil Rajput](https://github.com/nixrajput) diff --git a/docs/CDC.md b/docs/CDC.md new file mode 100644 index 0000000..bd381d4 --- /dev/null +++ b/docs/CDC.md @@ -0,0 +1,88 @@ +# Change data capture (CDC) + +CDC ("continuous sync" / follow mode) keeps a target database in step with a +source by streaming changes as they happen, rather than re-running a full +backup/restore. In siphon this is **scaffolding today** — the resume-state +plumbing and capability gating are in place, but continuous streaming is a +future deliverable and does not run. + +## Table of contents + +- [What exists](#what-exists) +- [The CLI surface](#the-cli-surface) +- [Status](#status) +- [Resume state](#resume-state) +- [Limitations](#limitations) + +## What exists + +`internal/app/cdc.go` holds: + +- **`CDCState`** — the persisted resume record (`job_id`, source/target + profiles, `last_applied_lsn` for Postgres, `last_binlog_file` + + `last_binlog_pos` for MySQL/MariaDB, `updated_at`). +- **`saveCDCState` / `loadCDCState`** — JSON state files under a per-user + directory resolved from `SIPHON_STATE_HOME`, then `XDG_STATE_HOME`, then + `$HOME/.local/state/siphon/cdc`. +- **`RunCDC`** — a capability-gated entry point. It resolves the source and + target profiles, then requires both drivers to advertise `CapCDC`. Its body is + a **polling scaffold**: a 10-second ticker that persists state each tick and + resumes from a prior run's state on restart. It is **not** real change + streaming. + +`CDCState` save/load round-trip, the state-dir resolution, and the +no-capability rejection are unit-tested (`internal/app/cdc_test.go`). + +## The CLI surface + +There is **no `siphon cdc` subcommand wired today.** `RunCDC` exists as an +application function but is not registered as a Cobra command +(`internal/cli/root.go` wires `backup`, `restore`, `sync`, `verify`, `inspect`, +`profile`, `dumps`, `config`, `schedule`, `tunnel` — no `cdc`). + +`sync --continuous` exposes the flag but does not run CDC; it returns a clear +error pointing at the (not-yet-wired) `siphon cdc` follow-up: + +```bash +siphon sync --from pg-prod --to pg-replica --continuous +# Error: continuous CDC sync is not wired here; use `siphon cdc` (Phase F Task 10) +``` + +## Status + +| Capability | Status | +| --- | --- | +| `CDCState` persistence (save/load) | ✅ Works (unit-tested) | +| State-dir resolution (`SIPHON_STATE_HOME` / `XDG_STATE_HOME`) | ✅ Works (unit-tested) | +| Resume from prior state on restart | ✅ Works (in the scaffold loop) | +| Capability gating on `CapCDC` | ✅ Works | +| `siphon cdc` CLI subcommand | ❌ Not wired (no Cobra command) | +| Continuous change streaming | ⚠️ Scaffold only (polling tick; not real CDC) | + +CDC does **not run today**. No driver declares `CapCDC` true, so `RunCDC` is +rejected with `ErrDriverUnsupported`. Even if a driver enabled it, the loop is a +polling scaffold, not logical-replication streaming. + +## Resume state + +When CDC streaming is built, `RunCDC` resumes from the last persisted position. +State files live at: + +```text +$SIPHON_STATE_HOME/cdc/.state # if SIPHON_STATE_HOME is set +$XDG_STATE_HOME/siphon/cdc/.state # else if XDG_STATE_HOME is set +$HOME/.local/state/siphon/cdc/.state # default +``` + +Each file is JSON: source/target profiles plus the last applied position (LSN +for Postgres, binlog file + offset for MySQL/MariaDB). + +## Limitations + +- **Not enabled on any driver** — `CapCDC` is `false` everywhere, so CDC is + rejected up front. +- **No real streaming** — the scaffold polls on a ticker. Real logical- + replication streaming (`pglogrepl` for Postgres, binlog tailing for + MySQL/MariaDB) is deferred to a future revision. +- **No CLI entry point** — `RunCDC` is internal application scaffolding, not a + user-facing command yet. diff --git a/docs/CROSS_ENGINE.md b/docs/CROSS_ENGINE.md new file mode 100644 index 0000000..ef19d8e --- /dev/null +++ b/docs/CROSS_ENGINE.md @@ -0,0 +1,99 @@ +# Cross-engine sync + +Cross-engine sync moves data between *different* database engines (e.g. Postgres +→ MySQL) by pivoting through a **canonical schema**: the source's native column +types are normalized into an engine-independent vocabulary, then mapped back to +the target engine's dialect. Rows travel as JSONL with per-engine identifier +quoting and bind-parameter placeholders. + +## Table of contents + +- [How it works](#how-it-works) +- [The CLI surface](#the-cli-surface) +- [Status](#status) +- [Type-mapping matrix](#type-mapping-matrix) +- [Scope and limitations](#scope-and-limitations) + +## How it works + +The canonical-schema machinery lives in `internal/app/canonical*.go`: + +- `CanonicalType` / `CanonicalColumn` / `CanonicalSchema` are the + engine-independent pivot (`canonical.go`). +- `MapToNative(engine, col)` renders a canonical column as a native column type + for `postgres`, `mysql`, or `mariadb`. `VARCHAR`/`NUMERIC` carry their + precision (and scale, for `NUMERIC`) when known; otherwise the type is emitted + bare and the engine applies its default. +- Identifiers are quoted per engine (`"…"` for Postgres, `` `…` `` for + MySQL/MariaDB) and escaped, since table/column names cannot be passed as bind + parameters. Placeholders are `$n` (Postgres) or `?` (MySQL/MariaDB). +- The emit/consume halves (`canonical_emit.go`, `canonical_consume.go`) stream a + schema + rows as JSONL. + +This machinery is built and unit-tested (`canonical_test.go`). + +## The CLI surface + +```bash +# Cross-engine sync from a Postgres source profile into a MySQL target profile. +siphon sync --from pg-prod --to mysql-staging --cross-engine +``` + +## Status + +| Capability | Status | +| --- | --- | +| Canonical type vocabulary + `MapToNative` matrix | ✅ Works (unit-tested) | +| JSONL emit/consume with per-engine quoting + placeholders | ✅ Works (unit-tested) | +| `sync --cross-engine` end-to-end | ⚠️ Gated off (follow-up) | + +The type-mapping and JSONL-transfer machinery exists and is tested, but +`sync --cross-engine` is **capability-gated** on `CrossEngineSource` +(source driver) and `CrossEngineTarget` (target driver). **No driver declares +either capability today**, so the request is rejected with +`ErrDriverUnsupported` (`internal/app/sync.go`, `runCrossEngineSync`). + +The reason is deliberate: cross-engine translation needs a *typed* +`CanonicalSchema` (column types), and `driver.Inspect` does not carry column +types yet. Rather than fabricate a schema, the path stays gated until typed +schema introspection lands and a driver flips its cross-engine capabilities to +true. At that point `runCrossEngineSync` gains the real +emit → translate → consume pipeline. + +```bash +siphon sync --from pg-prod --to mysql-staging --cross-engine +# Error: cross-engine sync requires typed schema introspection, not yet available +``` + +## Type-mapping matrix + +`MapToNative` maps each canonical type to a native type. These mappings are real +and usable even though the CLI path is gated: + +| Canonical type | Postgres | MySQL | MariaDB | +| --- | --- | --- | --- | +| `int` | `integer` | `INT` | `INT` | +| `bigint` | `bigint` | `BIGINT` | `BIGINT` | +| `text` | `text` | `TEXT` | `TEXT` | +| `varchar` | `varchar` | `VARCHAR` | `VARCHAR` | +| `boolean` | `boolean` | `TINYINT(1)` | `TINYINT(1)` | +| `numeric` | `numeric` | `DECIMAL` | `DECIMAL` | +| `uuid` | `uuid` | `CHAR(36)` | `CHAR(36)` | +| `timestamptz` | `timestamptz` | `TIMESTAMP` | `TIMESTAMP` | +| `json` | `jsonb` | `JSON` | `JSON` | + +`varchar` with a known length renders as e.g. `VARCHAR(255)`; `numeric` with +known precision/scale renders as e.g. `DECIMAL(10,2)`. + +## Scope and limitations + +The v1 cross-engine scope (once the path is wired) is **data only**: + +- Triggers, views, and stored functions are **skipped**. +- Index recreation is **not implemented yet**: the consume path issues + `CREATE TABLE IF NOT EXISTS` (column definitions only) plus row INSERTs, so + only data and table structure transfer. Indexes and any constraints beyond the + inline column defs are a follow-up. +- Foreign keys are **deferred**. +- Engines covered by the matrix: `postgres`, `mysql`, `mariadb`. An unknown + engine or an unmappable canonical type is an error from `MapToNative`. diff --git a/docs/INCREMENTAL.md b/docs/INCREMENTAL.md new file mode 100644 index 0000000..ca717b7 --- /dev/null +++ b/docs/INCREMENTAL.md @@ -0,0 +1,109 @@ +# Incremental backups + +Incremental backups capture only what changed since an earlier dump, so a chain +of small incrementals can stand in for repeated full dumps. siphon records the +relationship between dumps in the **dump envelope** (a 4 KB JSON header +prepended to every dump) and reconstructs the full picture at restore time by +walking the base → incremental chain. + +## Table of contents + +- [How it works](#how-it-works) +- [The CLI surface](#the-cli-surface) +- [Status](#status) +- [Examples](#examples) +- [Limitations and runtime gates](#limitations-and-runtime-gates) + +## How it works + +Every dump carries an envelope (`internal/dumps/envelope.go`) with a `type` +(`base` or `incremental`), a `base_id`/`parent_id`, and engine-specific resume +coordinates: `wal_start`/`wal_end` for Postgres, `binlog_file` + +`binlog_start`/`binlog_end` for MySQL/MariaDB. + +`Catalog.ResolveChain` (`internal/dumps/chain.go`) walks `parent_id` backwards +from a target dump to its base, detecting cycles and broken chains rather than +looping or silently truncating. `Restore` then applies the resolved chain in +order, base first (`internal/app/restore.go`). A plain, non-incremental dump +resolves to a single-element chain, so the same restore path serves both. + +The driver-level capture machinery also exists: + +- **Postgres** (`internal/driver/postgres/incremental.go`) creates a temporary + physical replication slot and records the start/end LSN around a base backup, + so a later incremental can resume from the correct WAL position. +- **MySQL/MariaDB** (`internal/driver/_mysqlcommon/incremental.go`) captures the + binlog file + position via `SHOW BINARY LOG STATUS` (MySQL 8.4+) or + `SHOW MASTER STATUS` (older MySQL / MariaDB). + +## The CLI surface + +```bash +# Restore a dump, walking its base→incremental chain automatically. +siphon restore --profile + +# Stop applying the chain after a specific dump (point-in-chain restore). +siphon restore --profile --up-to + +# Request an incremental backup (NOT yet wired — see Status). +siphon backup --incremental --base +``` + +## Status + +| Capability | Status | +| --- | --- | +| Dump envelope (type/base/parent, WAL & binlog fields) | ✅ Works | +| Chain resolution (`ResolveChain`) | ✅ Works | +| Chain-walking restore (base → incrementals, in order) | ✅ Works | +| `restore --up-to ` (stop chain early) | ✅ Works | +| Driver-level capture machinery (Postgres WAL slot/LSN, MySQL/MariaDB binlog pos) | ✅ Exists | +| `backup --incremental --base ` end-to-end | ⚠️ Not yet wired (follow-up) | + +The restore-side chain machinery and the envelope are in place and tested. The +`--incremental` backup path is a documented follow-up: the driver machinery and +the chain restore are both ready, but the wiring that ties them together — read +the base envelope for the parent WAL/binlog position, invoke the driver's +incremental capture method, and write an `incremental`-type catalog entry — is +not yet connected. Until then `siphon backup --incremental` returns a clear +error: *"incremental backup is not yet wired end-to-end (Phase F follow-up); the +driver-level machinery exists"* (`internal/cli/backup.go`). + +## Examples + +Chain-walking restore (works today). If `inc-2` was built on `inc-1` on +`base-0`, restoring `inc-2` applies all three in order: + +```bash +siphon restore inc-2 --profile prod-replica +``` + +Point-in-chain restore with `--up-to` (works today) — apply only up to and +including `inc-1`, skipping `inc-2`: + +```bash +siphon restore inc-2 --profile prod-replica --up-to inc-1 +``` + +A typo'd `--up-to` is rejected (the dump isn't in the chain) rather than +silently restoring more than asked. + +Requesting an incremental backup (returns the not-wired error today): + +```bash +siphon backup prod --incremental --base base-0 +# Error: incremental backup is not yet wired end-to-end (Phase F follow-up); +# the driver-level machinery exists +``` + +## Limitations and runtime gates + +These apply to the incremental **backup** path once it is wired: + +- **Postgres** anchors WAL retention with a temporary physical replication slot. + Orphan-slot cleanup is **not yet built** — an orphaned slot pins WAL on the + server and must be dropped manually until automatic cleanup ships. +- **MySQL/MariaDB** require `binlog_format=ROW` for usable incrementals. +- **Cross-version incrementals are unsupported** (`CrossVersionIncremental: + false`): a chain must be captured and restored against the same engine major + version. diff --git a/internal/app/backup.go b/internal/app/backup.go index c4fa215..3751879 100644 --- a/internal/app/backup.go +++ b/internal/app/backup.go @@ -78,6 +78,17 @@ func Backup(parent context.Context, d Deps, opt BackupOpts) (<-chan jobs.Event, h := sha256.New() tee := io.MultiWriter(f, h) + env := &dumps.Envelope{ + Type: dumps.EnvelopeBase, + Driver: resolved.Driver, + Created: time.Now().UTC(), + } + if _, err := dumps.WriteEnvelope(tee, env); err != nil { + _ = f.Close() + _ = os.Remove(tmpPath) + return err + } + emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "dumping"}) backupErr := conn.Backup(ctx, driver.BackupOpts{ diff --git a/internal/app/canonical.go b/internal/app/canonical.go new file mode 100644 index 0000000..5903768 --- /dev/null +++ b/internal/app/canonical.go @@ -0,0 +1,140 @@ +package app + +import ( + "fmt" + "strconv" + "strings" +) + +// CanonicalType is the engine-independent column type vocabulary. A source +// engine's native types are normalized into these; a target engine maps them +// back to its own dialect via MapToNative. This is the pivot that lets data +// move between Postgres, MySQL, and MariaDB. +type CanonicalType string + +const ( + CTInt CanonicalType = "int" + CTBigInt CanonicalType = "bigint" + CTText CanonicalType = "text" + CTVarchar CanonicalType = "varchar" + CTBoolean CanonicalType = "boolean" + CTNumeric CanonicalType = "numeric" + CTUUID CanonicalType = "uuid" + CTTimestampTZ CanonicalType = "timestamptz" + CTJSON CanonicalType = "json" +) + +// CanonicalColumn describes one column in canonical terms. +// +// Precision/Scale are only meaningful for CTVarchar (Precision = length) and +// CTNumeric (Precision = total digits, Scale = fractional digits). When both +// are zero the type is emitted bare (e.g. VARCHAR, DECIMAL) — acceptable for +// v1; engines apply their own defaults. +type CanonicalColumn struct { + Name string `json:"name"` + Type CanonicalType `json:"type"` + Nullable bool `json:"nullable"` + Precision int `json:"precision,omitempty"` + Scale int `json:"scale,omitempty"` +} + +// CanonicalTable is a table name plus its ordered columns. +type CanonicalTable struct { + Name string `json:"name"` + Columns []CanonicalColumn `json:"columns"` +} + +// CanonicalSchema is the full set of tables to transfer. +type CanonicalSchema struct { + Tables []CanonicalTable `json:"tables"` +} + +// CanonicalRow is one row of data, keyed by column name. The compact JSON keys +// keep the JSONL stream small across many rows. +type CanonicalRow struct { + Table string `json:"t"` + Values map[string]any `json:"v"` +} + +// MapToNative renders a canonical column as a native column type for the given +// target engine. Returns an error for an unknown engine or an unmappable type. +// +// VARCHAR/NUMERIC with Precision>0 carry their precision (and Scale for +// NUMERIC) into the rendered type; otherwise the type is emitted bare. +func MapToNative(engine string, col CanonicalColumn) (string, error) { + var m map[CanonicalType]string + switch engine { + case "postgres": + m = map[CanonicalType]string{ + CTInt: "integer", + CTBigInt: "bigint", + CTText: "text", + CTVarchar: "varchar", + CTBoolean: "boolean", + CTNumeric: "numeric", + CTUUID: "uuid", + CTTimestampTZ: "timestamptz", + CTJSON: "jsonb", + } + case "mysql", "mariadb": + m = map[CanonicalType]string{ + CTInt: "INT", + CTBigInt: "BIGINT", + CTText: "TEXT", + CTVarchar: "VARCHAR", + CTBoolean: "TINYINT(1)", + CTNumeric: "DECIMAL", + CTUUID: "CHAR(36)", + CTTimestampTZ: "TIMESTAMP", + CTJSON: "JSON", + } + default: + return "", fmt.Errorf("cross-engine: unknown engine %q", engine) + } + + native, ok := m[col.Type] + if !ok { + return "", fmt.Errorf("cross-engine: engine %q has no mapping for canonical type %q", engine, col.Type) + } + + // Decorate variable-length / fixed-precision types when precision is known. + switch col.Type { + case CTVarchar: + if col.Precision > 0 { + native = fmt.Sprintf("%s(%d)", native, col.Precision) + } + case CTNumeric: + if col.Precision > 0 { + native = fmt.Sprintf("%s(%d,%d)", native, col.Precision, col.Scale) + } + } + return native, nil +} + +// quoteIdent quotes a SQL identifier for the given engine, escaping any quote +// characters inside it. Identifiers (table/column names) CANNOT be passed as +// bound parameters, so quoting+escaping per engine is the mitigation against a +// name that contains a space, keyword, quote, or `;` — whether malicious or +// merely awkward. Every identifier that reaches generated SQL must pass through +// here. +func quoteIdent(engine, ident string) (string, error) { + switch engine { + case "postgres": + return `"` + strings.ReplaceAll(ident, `"`, `""`) + `"`, nil + case "mysql", "mariadb": + return "`" + strings.ReplaceAll(ident, "`", "``") + "`", nil + default: + return "", fmt.Errorf("cross-engine: unknown engine %q", engine) + } +} + +// placeholder returns the bind-parameter placeholder for position n (1-based) +// in the given engine's dialect: $n for Postgres, ? for MySQL/MariaDB. +func placeholder(engine string, n int) string { + switch engine { + case "postgres": + return "$" + strconv.Itoa(n) + default: + return "?" + } +} diff --git a/internal/app/canonical_consume.go b/internal/app/canonical_consume.go new file mode 100644 index 0000000..ae9ca80 --- /dev/null +++ b/internal/app/canonical_consume.go @@ -0,0 +1,142 @@ +package app + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "io" + "strings" +) + +// ConsumeCanonical reads a stream produced by EmitCanonical and replays it into +// db using engine's dialect: the first JSON value is the schema header (used to +// CREATE TABLE IF NOT EXISTS for each table), every subsequent JSON value is a +// CanonicalRow that is INSERTed. Values are always bound as parameters; only +// identifiers are interpolated (quoted via quoteIdent). +// +// A json.Decoder reads successive JSON values directly, so there is no +// token-size cap: a single large row can no longer abort the stream mid-replay +// (which a bufio.Scanner's "token too long" would have done). +func ConsumeCanonical(ctx context.Context, db *sql.DB, engine string, r io.Reader) error { + dec := json.NewDecoder(r) + + var header struct { + Schema *CanonicalSchema `json:"schema"` + } + if err := dec.Decode(&header); err != nil { + if errors.Is(err, io.EOF) { + return fmt.Errorf("cross-engine: empty stream, schema header missing") + } + return fmt.Errorf("cross-engine: decode schema header: %w", err) + } + if header.Schema == nil { + return fmt.Errorf("cross-engine: schema header missing") + } + + if err := synthCreateTables(ctx, db, engine, header.Schema); err != nil { + return err + } + + for { + var row CanonicalRow + if err := dec.Decode(&row); err != nil { + if errors.Is(err, io.EOF) { + break + } + return fmt.Errorf("cross-engine: decode row: %w", err) + } + if err := insertRow(ctx, db, engine, row); err != nil { + return err + } + } + return nil +} + +// synthCreateTables issues a CREATE TABLE IF NOT EXISTS per table. +func synthCreateTables(ctx context.Context, db *sql.DB, engine string, schema *CanonicalSchema) error { + for _, t := range schema.Tables { + stmt, err := buildCreateTableSQL(engine, t) + if err != nil { + return err + } + if _, err := db.ExecContext(ctx, stmt); err != nil { + return fmt.Errorf("cross-engine: create table %s: %w", t.Name, err) + } + } + return nil +} + +// insertRow inserts a single CanonicalRow, binding values as parameters. +func insertRow(ctx context.Context, db *sql.DB, engine string, row CanonicalRow) error { + // Build column list and value list in the SAME pass so col[i] and val[i] + // stay aligned even though map iteration order is unspecified. + cols := make([]string, 0, len(row.Values)) + vals := make([]any, 0, len(row.Values)) + for c, v := range row.Values { + cols = append(cols, c) + vals = append(vals, v) + } + + stmt, err := buildInsertSQL(engine, row.Table, cols) + if err != nil { + return err + } + if _, err := db.ExecContext(ctx, stmt, vals...); err != nil { + return fmt.Errorf("cross-engine: insert into %s: %w", row.Table, err) + } + return nil +} + +// buildCreateTableSQL renders a CREATE TABLE IF NOT EXISTS statement with every +// identifier quoted for the engine and each column mapped via MapToNative. Pure +// function — unit-testable without a DB. +func buildCreateTableSQL(engine string, t CanonicalTable) (string, error) { + qt, err := quoteIdent(engine, t.Name) + if err != nil { + return "", err + } + defs := make([]string, len(t.Columns)) + for i, c := range t.Columns { + qc, err := quoteIdent(engine, c.Name) + if err != nil { + return "", err + } + native, err := MapToNative(engine, c) + if err != nil { + return "", err + } + def := qc + " " + native + if !c.Nullable { + def += " NOT NULL" + } + defs[i] = def + } + return fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s)", qt, strings.Join(defs, ", ")), nil +} + +// buildInsertSQL renders an INSERT with quoted identifiers and per-engine +// placeholders. Pure function — unit-testable without a DB. Returns an error +// for an empty column set (an INSERT with no columns is malformed). +func buildInsertSQL(engine, table string, cols []string) (string, error) { + if len(cols) == 0 { + return "", fmt.Errorf("cross-engine: insert into %s: no columns", table) + } + qt, err := quoteIdent(engine, table) + if err != nil { + return "", err + } + qcols := make([]string, len(cols)) + phs := make([]string, len(cols)) + for i, c := range cols { + qc, err := quoteIdent(engine, c) + if err != nil { + return "", err + } + qcols[i] = qc + phs[i] = placeholder(engine, i+1) + } + return fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", + qt, strings.Join(qcols, ","), strings.Join(phs, ",")), nil +} diff --git a/internal/app/canonical_emit.go b/internal/app/canonical_emit.go new file mode 100644 index 0000000..b1a01f8 --- /dev/null +++ b/internal/app/canonical_emit.go @@ -0,0 +1,107 @@ +package app + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "io" + "strings" +) + +// EmitCanonical writes a table-by-table snapshot of schema as JSONL to w. +// +// Line 1 is the schema header ({"schema": {...}}); each subsequent line is one +// CanonicalRow ({"t": table, "v": {col: val}}). The engine param names the +// SOURCE engine and is used only to quote identifiers in the SELECT — values +// are never interpolated. +func EmitCanonical(ctx context.Context, db *sql.DB, engine string, schema *CanonicalSchema, w io.Writer) error { + if err := writeJSONL(w, map[string]*CanonicalSchema{"schema": schema}); err != nil { + return err + } + for _, t := range schema.Tables { + if err := emitTable(ctx, db, engine, t, w); err != nil { + return err + } + } + return nil +} + +// emitTable streams one table's rows as CanonicalRow JSONL lines. +func emitTable(ctx context.Context, db *sql.DB, engine string, t CanonicalTable, w io.Writer) error { + query, err := buildSelectSQL(engine, t) + if err != nil { + return err + } + + rows, err := db.QueryContext(ctx, query) + if err != nil { + return fmt.Errorf("cross-engine: select %s: %w", t.Name, err) + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + vals := make([]any, len(t.Columns)) + ptrs := make([]any, len(t.Columns)) + for i := range vals { + ptrs[i] = &vals[i] + } + if err := rows.Scan(ptrs...); err != nil { + return fmt.Errorf("cross-engine: scan %s: %w", t.Name, err) + } + m := make(map[string]any, len(t.Columns)) + for i, c := range t.Columns { + m[c.Name] = normalizeScanned(vals[i]) + } + if err := writeJSONL(w, CanonicalRow{Table: t.Name, Values: m}); err != nil { + return err + } + } + if err := rows.Err(); err != nil { + return fmt.Errorf("cross-engine: iterate %s: %w", t.Name, err) + } + return nil +} + +// buildSelectSQL builds `SELECT FROM ` with every +// identifier quoted for the engine. Pure function — unit-testable without a DB. +func buildSelectSQL(engine string, t CanonicalTable) (string, error) { + qt, err := quoteIdent(engine, t.Name) + if err != nil { + return "", err + } + cols := make([]string, len(t.Columns)) + for i, c := range t.Columns { + qc, err := quoteIdent(engine, c.Name) + if err != nil { + return "", err + } + cols[i] = qc + } + return fmt.Sprintf("SELECT %s FROM %s", strings.Join(cols, ", "), qt), nil +} + +// normalizeScanned converts a value scanned from database/sql into a form that +// round-trips correctly through JSON. database/sql commonly yields []byte for +// text/numeric/json columns; marshaling []byte to JSON produces base64, which +// corrupts the value when ConsumeCanonical re-inserts it. Converting to string +// makes it marshal as a JSON string instead, preserving the original bytes. +func normalizeScanned(v any) any { + if b, ok := v.([]byte); ok { + return string(b) + } + return v +} + +// writeJSONL marshals v and writes it followed by a newline. +func writeJSONL(w io.Writer, v any) error { + b, err := json.Marshal(v) + if err != nil { + return err + } + if _, err := w.Write(b); err != nil { + return err + } + _, err = w.Write([]byte("\n")) + return err +} diff --git a/internal/app/canonical_test.go b/internal/app/canonical_test.go new file mode 100644 index 0000000..438a3e7 --- /dev/null +++ b/internal/app/canonical_test.go @@ -0,0 +1,407 @@ +package app + +import ( + "encoding/json" + "strings" + "testing" +) + +// --- MapToNative ----------------------------------------------------------- + +func TestMapToNative_CoreTypes(t *testing.T) { + coreTypes := []CanonicalType{ + CTInt, CTBigInt, CTText, CTBoolean, CTUUID, + CTVarchar, CTNumeric, CTTimestampTZ, CTJSON, + } + for _, engine := range []string{"postgres", "mysql", "mariadb"} { + for _, ct := range coreTypes { + native, err := MapToNative(engine, CanonicalColumn{Name: "c", Type: ct}) + if err != nil { + t.Errorf("MapToNative(%q, %q): unexpected error: %v", engine, ct, err) + } + if native == "" { + t.Errorf("MapToNative(%q, %q): empty native type", engine, ct) + } + } + } +} + +func TestMapToNative_UnknownEngine(t *testing.T) { + if _, err := MapToNative("oracle", CanonicalColumn{Type: CTInt}); err == nil { + t.Fatal("MapToNative with unknown engine: want error, got nil") + } +} + +func TestMapToNative_UnknownType(t *testing.T) { + if _, err := MapToNative("postgres", CanonicalColumn{Type: CanonicalType("geometry")}); err == nil { + t.Fatal("MapToNative with unknown type: want error, got nil") + } +} + +func TestMapToNative_PrecisionDecoration(t *testing.T) { + // VARCHAR with length. + got, err := MapToNative("postgres", CanonicalColumn{Type: CTVarchar, Precision: 255}) + if err != nil { + t.Fatal(err) + } + if got != "varchar(255)" { + t.Errorf("varchar(255): got %q", got) + } + // NUMERIC with precision+scale (MySQL DECIMAL). + got, err = MapToNative("mysql", CanonicalColumn{Type: CTNumeric, Precision: 10, Scale: 2}) + if err != nil { + t.Fatal(err) + } + if got != "DECIMAL(10,2)" { + t.Errorf("DECIMAL(10,2): got %q", got) + } + // Bare when precision is zero (documented v1 behavior). + got, err = MapToNative("postgres", CanonicalColumn{Type: CTVarchar}) + if err != nil { + t.Fatal(err) + } + if got != "varchar" { + t.Errorf("bare varchar: got %q", got) + } +} + +// --- quoteIdent (INJECTION GUARD) ------------------------------------------ + +func TestQuoteIdent_Postgres(t *testing.T) { + got, err := quoteIdent("postgres", "users") + if err != nil { + t.Fatal(err) + } + if got != `"users"` { + t.Errorf("postgres quoteIdent: got %q want %q", got, `"users"`) + } +} + +func TestQuoteIdent_PostgresEscapesQuote(t *testing.T) { + got, err := quoteIdent("postgres", `we"ird`) + if err != nil { + t.Fatal(err) + } + if got != `"we""ird"` { + t.Errorf("postgres quote-doubling: got %q want %q", got, `"we""ird"`) + } +} + +func TestQuoteIdent_MySQL(t *testing.T) { + got, err := quoteIdent("mysql", "users") + if err != nil { + t.Fatal(err) + } + if got != "`users`" { + t.Errorf("mysql quoteIdent: got %q want %q", got, "`users`") + } +} + +func TestQuoteIdent_MySQLEscapesBacktick(t *testing.T) { + got, err := quoteIdent("mysql", "we`ird") + if err != nil { + t.Fatal(err) + } + if got != "`we``ird`" { + t.Errorf("mysql backtick-doubling: got %q want %q", got, "`we``ird`") + } +} + +func TestQuoteIdent_UnknownEngine(t *testing.T) { + if _, err := quoteIdent("oracle", "users"); err == nil { + t.Fatal("quoteIdent with unknown engine: want error, got nil") + } +} + +// TestQuoteIdent_InjectionIsNeutralized is the core security assertion: a +// malicious identifier is wrapped+escaped, never passed through raw, so the +// embedded statement terminator and quote cannot break out of the identifier. +func TestQuoteIdent_InjectionIsNeutralized(t *testing.T) { + evil := `x"; DROP TABLE y; --` + + pg, err := quoteIdent("postgres", evil) + if err != nil { + t.Fatal(err) + } + // The inner " must be doubled, and the whole thing wrapped in quotes. + want := `"x""; DROP TABLE y; --"` + if pg != want { + t.Errorf("postgres injection: got %q want %q", pg, want) + } + // The raw (un-doubled) attacker substring must NOT appear verbatim. + if strings.Contains(pg, `x"; DROP`) { + t.Errorf("postgres injection: raw quote leaked through: %q", pg) + } + + my, err := quoteIdent("mysql", "x`; DROP TABLE y; --") + if err != nil { + t.Fatal(err) + } + wantMy := "`x``; DROP TABLE y; --`" + if my != wantMy { + t.Errorf("mysql injection: got %q want %q", my, wantMy) + } +} + +// --- placeholder ----------------------------------------------------------- + +func TestPlaceholder(t *testing.T) { + cases := []struct { + engine string + n int + want string + }{ + {"postgres", 1, "$1"}, + {"postgres", 3, "$3"}, + {"mysql", 1, "?"}, + {"mariadb", 2, "?"}, + } + for _, c := range cases { + if got := placeholder(c.engine, c.n); got != c.want { + t.Errorf("placeholder(%q, %d): got %q want %q", c.engine, c.n, got, c.want) + } + } +} + +// --- SQL builders (pure, DB-free) ------------------------------------------ + +func twoColTable() CanonicalTable { + return CanonicalTable{ + Name: "t", + Columns: []CanonicalColumn{ + {Name: "id", Type: CTInt, Nullable: false}, + {Name: "name", Type: CTText, Nullable: true}, + }, + } +} + +func TestBuildCreateTableSQL_Postgres(t *testing.T) { + got, err := buildCreateTableSQL("postgres", twoColTable()) + if err != nil { + t.Fatal(err) + } + want := `CREATE TABLE IF NOT EXISTS "t" ("id" integer NOT NULL, "name" text)` + if got != want { + t.Errorf("postgres CREATE:\n got %q\nwant %q", got, want) + } +} + +func TestBuildCreateTableSQL_MySQL(t *testing.T) { + got, err := buildCreateTableSQL("mysql", twoColTable()) + if err != nil { + t.Fatal(err) + } + want := "CREATE TABLE IF NOT EXISTS `t` (`id` INT NOT NULL, `name` TEXT)" + if got != want { + t.Errorf("mysql CREATE:\n got %q\nwant %q", got, want) + } +} + +func TestBuildCreateTableSQL_UnknownEngine(t *testing.T) { + if _, err := buildCreateTableSQL("oracle", twoColTable()); err == nil { + t.Fatal("buildCreateTableSQL unknown engine: want error, got nil") + } +} + +// TestBuildCreateTableSQL_InjectionGuard proves a hostile column name flows +// through the builder quoted+escaped, not breaking out of the DDL. +func TestBuildCreateTableSQL_InjectionGuard(t *testing.T) { + tbl := CanonicalTable{ + Name: "t", + Columns: []CanonicalColumn{{Name: `evil"col`, Type: CTInt, Nullable: true}}, + } + got, err := buildCreateTableSQL("postgres", tbl) + if err != nil { + t.Fatal(err) + } + want := `CREATE TABLE IF NOT EXISTS "t" ("evil""col" integer)` + if got != want { + t.Errorf("injection-guarded CREATE:\n got %q\nwant %q", got, want) + } + if strings.Contains(got, `evil"col" integer)`) && !strings.Contains(got, `evil""col`) { + t.Errorf("raw quote leaked into DDL: %q", got) + } +} + +func TestBuildInsertSQL_Postgres(t *testing.T) { + got, err := buildInsertSQL("postgres", "t", []string{"id", "name"}) + if err != nil { + t.Fatal(err) + } + want := `INSERT INTO "t" ("id","name") VALUES ($1,$2)` + if got != want { + t.Errorf("postgres INSERT:\n got %q\nwant %q", got, want) + } +} + +func TestBuildInsertSQL_MySQL(t *testing.T) { + got, err := buildInsertSQL("mysql", "t", []string{"id", "name"}) + if err != nil { + t.Fatal(err) + } + want := "INSERT INTO `t` (`id`,`name`) VALUES (?,?)" + if got != want { + t.Errorf("mysql INSERT:\n got %q\nwant %q", got, want) + } +} + +func TestBuildInsertSQL_EmptyColumns(t *testing.T) { + if _, err := buildInsertSQL("postgres", "t", nil); err == nil { + t.Fatal("buildInsertSQL with no columns: want error, got nil") + } +} + +func TestBuildInsertSQL_UnknownEngine(t *testing.T) { + if _, err := buildInsertSQL("oracle", "t", []string{"id"}); err == nil { + t.Fatal("buildInsertSQL unknown engine: want error, got nil") + } +} + +func TestBuildSelectSQL_Postgres(t *testing.T) { + got, err := buildSelectSQL("postgres", twoColTable()) + if err != nil { + t.Fatal(err) + } + want := `SELECT "id", "name" FROM "t"` + if got != want { + t.Errorf("postgres SELECT:\n got %q\nwant %q", got, want) + } +} + +func TestBuildSelectSQL_MySQL(t *testing.T) { + got, err := buildSelectSQL("mysql", twoColTable()) + if err != nil { + t.Fatal(err) + } + want := "SELECT `id`, `name` FROM `t`" + if got != want { + t.Errorf("mysql SELECT:\n got %q\nwant %q", got, want) + } +} + +// --- JSONL framing --------------------------------------------------------- + +func TestWriteJSONL_SchemaHeaderFirst(t *testing.T) { + var sb strings.Builder + schema := &CanonicalSchema{Tables: []CanonicalTable{twoColTable()}} + if err := writeJSONL(&sb, map[string]*CanonicalSchema{"schema": schema}); err != nil { + t.Fatal(err) + } + out := sb.String() + if !strings.HasPrefix(out, `{"schema":`) { + t.Errorf("first line must start with schema key, got %q", out) + } + if !strings.HasSuffix(out, "\n") { + t.Error("writeJSONL must terminate the line with a newline") + } +} + +func TestCanonicalRow_JSONRoundTrip(t *testing.T) { + orig := CanonicalRow{ + Table: "users", + Values: map[string]any{ + "id": float64(7), // JSON numbers decode as float64 + "name": "alice", + "ok": true, + "nil": nil, + }, + } + b, err := json.Marshal(orig) + if err != nil { + t.Fatal(err) + } + // Compact keys keep the stream small. + if !strings.Contains(string(b), `"t":"users"`) || !strings.Contains(string(b), `"v":`) { + t.Errorf("CanonicalRow uses compact keys t/v, got %s", b) + } + var back CanonicalRow + if err := json.Unmarshal(b, &back); err != nil { + t.Fatal(err) + } + if back.Table != orig.Table { + t.Errorf("table mismatch: got %q want %q", back.Table, orig.Table) + } + if len(back.Values) != len(orig.Values) { + t.Fatalf("values length mismatch: got %d want %d", len(back.Values), len(orig.Values)) + } + for k, v := range orig.Values { + if back.Values[k] != v { + t.Errorf("value %q mismatch: got %v want %v", k, back.Values[k], v) + } + } +} + +// TestSchemaHeader_RoundTrip confirms a schema header marshals and unmarshals +// through the same envelope ConsumeCanonical expects. +func TestSchemaHeader_RoundTrip(t *testing.T) { + schema := &CanonicalSchema{Tables: []CanonicalTable{ + {Name: "users", Columns: []CanonicalColumn{ + {Name: "id", Type: CTBigInt, Nullable: false}, + {Name: "email", Type: CTVarchar, Nullable: true, Precision: 320}, + }}, + }} + var sb strings.Builder + if err := writeJSONL(&sb, map[string]*CanonicalSchema{"schema": schema}); err != nil { + t.Fatal(err) + } + var header struct { + Schema *CanonicalSchema `json:"schema"` + } + if err := json.Unmarshal([]byte(strings.TrimSpace(sb.String())), &header); err != nil { + t.Fatal(err) + } + if header.Schema == nil || len(header.Schema.Tables) != 1 { + t.Fatalf("schema header did not round-trip: %+v", header.Schema) + } + got := header.Schema.Tables[0] + if got.Name != "users" || len(got.Columns) != 2 || got.Columns[1].Precision != 320 { + t.Errorf("schema fields lost in round-trip: %+v", got) + } +} + +// --- normalizeScanned ------------------------------------------------------ + +// TestNormalizeScanned_BytesBecomeString proves a []byte value scanned from +// database/sql is converted to a string so it marshals as a JSON string and +// round-trips intact, rather than base64-encoding (which would corrupt the +// value before ConsumeCanonical re-inserts it). +func TestNormalizeScanned_BytesBecomeString(t *testing.T) { + in := []byte("hello-world") + got := normalizeScanned(in) + s, ok := got.(string) + if !ok { + t.Fatalf("normalizeScanned([]byte) returned %T; want string", got) + } + if s != "hello-world" { + t.Fatalf("normalizeScanned = %q; want %q", s, "hello-world") + } + + // Round-trip through the row marshal: a []byte value must come back as the + // original text, NOT base64. + row := CanonicalRow{Table: "t", Values: map[string]any{"c": normalizeScanned([]byte("café"))}} + b, err := json.Marshal(row) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(b), "Y2Fmw") { // base64 prefix of "café" would leak in + t.Fatalf("[]byte was base64-encoded, not stringified: %s", b) + } + var back CanonicalRow + if err := json.Unmarshal(b, &back); err != nil { + t.Fatal(err) + } + if back.Values["c"] != "café" { + t.Fatalf("round-trip = %v; want %q", back.Values["c"], "café") + } +} + +// TestNormalizeScanned_NonBytesUnchanged confirms non-[]byte values pass +// through untouched. +func TestNormalizeScanned_NonBytesUnchanged(t *testing.T) { + if got := normalizeScanned(int64(42)); got != int64(42) { + t.Fatalf("normalizeScanned(int64) = %v; want 42", got) + } + if got := normalizeScanned(nil); got != nil { + t.Fatalf("normalizeScanned(nil) = %v; want nil", got) + } +} diff --git a/internal/app/cdc.go b/internal/app/cdc.go new file mode 100644 index 0000000..891454b --- /dev/null +++ b/internal/app/cdc.go @@ -0,0 +1,110 @@ +package app + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "time" + + "github.com/nixrajput/siphon/internal/jobs" +) + +// CDCState is persisted between runs so a long-running continuous sync can +// resume from the last applied position after a restart. +type CDCState struct { + JobID string `json:"job_id"` + Source string `json:"source_profile"` + Target string `json:"target_profile"` + LastAppliedLSN string `json:"last_applied_lsn,omitempty"` + LastBinlogFile string `json:"last_binlog_file,omitempty"` + LastBinlogPos uint64 `json:"last_binlog_pos,omitempty"` + UpdatedAt time.Time `json:"updated_at"` +} + +// CDCStateDir returns the per-user directory holding CDC resume state. It +// honors SIPHON_STATE_HOME, then XDG_STATE_HOME, then $HOME/.local/state — +// mirroring how internal/config resolves its config path, so tests can redirect +// it without writing to the real home. +func CDCStateDir() string { + if v := os.Getenv("SIPHON_STATE_HOME"); v != "" { + return filepath.Join(v, "cdc") + } + if v := os.Getenv("XDG_STATE_HOME"); v != "" { + return filepath.Join(v, "siphon", "cdc") + } + home, _ := os.UserHomeDir() + return filepath.Join(home, ".local", "state", "siphon", "cdc") +} + +func saveCDCState(s *CDCState) error { + if err := os.MkdirAll(CDCStateDir(), 0o700); err != nil { + return err + } + s.UpdatedAt = time.Now().UTC() + body, err := json.MarshalIndent(s, "", " ") + if err != nil { + return err + } + return os.WriteFile(filepath.Join(CDCStateDir(), s.JobID+".state"), body, 0o600) +} + +func loadCDCState(jobID string) (*CDCState, error) { + body, err := os.ReadFile(filepath.Join(CDCStateDir(), jobID+".state")) + if err != nil { + return nil, err + } + s := &CDCState{} + return s, json.Unmarshal(body, s) +} + +// RunCDC starts (or resumes) a continuous sync. It is capability-gated: both +// the source and target driver must advertise CapCDC. No driver does today +// (CDC streaming via logical replication / binlog tailing is a Phase F +// follow-up), so this returns ErrDriverUnsupported — the honest scaffold state. +// When a driver gains CDC support, the polling loop below is replaced by real +// logical-replication streaming (pglogrepl for Postgres). +func RunCDC(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, string, error) { + if _, err := d.Profiles.Resolve(opt.From); err != nil { + return nil, "", err + } + if _, err := d.Profiles.Resolve(opt.To); err != nil { + return nil, "", err + } + if err := RequireCapability(d, opt.From, CapCDC); err != nil { + return nil, "", err + } + if err := RequireCapability(d, opt.To, CapCDC); err != nil { + return nil, "", err + } + + return d.Runner.Run(parent, jobs.Job{ + Stage: "cdc", + Func: func(ctx context.Context, emit func(jobs.Event)) error { + emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "CDC mode started"}) + // Phase F scaffold: a polling heartbeat that persists state each tick. + // A future revision replaces this with real logical-replication + // streaming (pglogrepl for Postgres, binlog tailing for MySQL/MariaDB). + const jobID = "cdc" + state := &CDCState{JobID: jobID, Source: opt.From, Target: opt.To} + // Resume from a prior run's persisted position when one exists. + if prev, err := loadCDCState(jobID); err == nil { + state = prev + } + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + _ = saveCDCState(state) + return ctx.Err() + case <-ticker.C: + if err := saveCDCState(state); err != nil { + return err + } + emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "CDC tick"}) + } + } + }, + }) +} diff --git a/internal/app/cdc_test.go b/internal/app/cdc_test.go new file mode 100644 index 0000000..5c39b55 --- /dev/null +++ b/internal/app/cdc_test.go @@ -0,0 +1,75 @@ +package app + +import ( + "context" + "errors" + "path/filepath" + "strings" + "testing" + + "github.com/nixrajput/siphon/internal/driver" + "github.com/nixrajput/siphon/internal/errs" + "github.com/nixrajput/siphon/internal/jobs" +) + +func TestCDCState_SaveLoad_Roundtrip(t *testing.T) { + t.Setenv("SIPHON_STATE_HOME", t.TempDir()) + + in := &CDCState{JobID: "job1", Source: "a", Target: "b", LastAppliedLSN: "0/16B3748"} + if err := saveCDCState(in); err != nil { + t.Fatalf("saveCDCState: %v", err) + } + + out, err := loadCDCState("job1") + if err != nil { + t.Fatalf("loadCDCState: %v", err) + } + if out.Source != "a" || out.Target != "b" { + t.Fatalf("Source/Target = %q/%q; want a/b", out.Source, out.Target) + } + if out.LastAppliedLSN != "0/16B3748" { + t.Fatalf("LastAppliedLSN = %q; want 0/16B3748", out.LastAppliedLSN) + } + if out.UpdatedAt.IsZero() { + t.Fatal("UpdatedAt is zero; saveCDCState should stamp it") + } +} + +func TestCDCStateDir_HonorsEnv(t *testing.T) { + t.Run("SIPHON_STATE_HOME", func(t *testing.T) { + tmp := t.TempDir() + t.Setenv("SIPHON_STATE_HOME", tmp) + t.Setenv("XDG_STATE_HOME", "") + if got, want := CDCStateDir(), filepath.Join(tmp, "cdc"); got != want { + t.Fatalf("CDCStateDir() = %q; want %q", got, want) + } + }) + + t.Run("XDG_STATE_HOME", func(t *testing.T) { + tmp := t.TempDir() + t.Setenv("SIPHON_STATE_HOME", "") + t.Setenv("XDG_STATE_HOME", tmp) + got := CDCStateDir() + if want := filepath.Join(tmp, "siphon", "cdc"); got != want { + t.Fatalf("CDCStateDir() = %q; want %q", got, want) + } + if !strings.HasSuffix(got, filepath.Join("siphon", "cdc")) { + t.Fatalf("CDCStateDir() = %q; want suffix siphon/cdc", got) + } + }) +} + +func TestRunCDC_RejectsWithoutCapability(t *testing.T) { + // A driver with CDC:false (and a Runner so the call is well-formed even + // though the cap gate rejects before the job ever runs). + deps := capDeps(driver.Capabilities{CDC: false}) + deps.Runner = jobs.NewRunner() + + _, _, err := RunCDC(context.Background(), deps, SyncOpts{From: "p", To: "p"}) + if err == nil { + t.Fatal("RunCDC = nil; want error (no driver advertises CDC)") + } + if !errors.Is(err, errs.ErrDriverUnsupported) { + t.Fatalf("errors.Is(err, ErrDriverUnsupported) = false; err = %v", err) + } +} diff --git a/internal/app/restore.go b/internal/app/restore.go index 6b5f760..fe7ba89 100644 --- a/internal/app/restore.go +++ b/internal/app/restore.go @@ -5,6 +5,8 @@ import ( "os" "github.com/nixrajput/siphon/internal/driver" + "github.com/nixrajput/siphon/internal/dumps" + "github.com/nixrajput/siphon/internal/errs" "github.com/nixrajput/siphon/internal/jobs" ) @@ -16,9 +18,12 @@ type RestoreOpts struct { SchemaOnly bool DataOnly bool Clean bool + UpTo string // optional: stop applying the chain after this dump ID } -// Restore replays a dump from the catalog into the target profile. +// Restore resolves the dump's chain (base + incrementals) and applies it in +// order into the target profile. For a plain (non-incremental) dump the chain +// is a single element. --up-to stops the chain early at the named dump. func Restore(parent context.Context, d Deps, opt RestoreOpts) (<-chan jobs.Event, string, error) { resolved, err := d.Profiles.Resolve(opt.Profile) if err != nil { @@ -29,6 +34,17 @@ func Restore(parent context.Context, d Deps, opt RestoreOpts) (<-chan jobs.Event return nil, "", err } + chain, err := d.Dumps.ResolveChain(opt.DumpID) + if err != nil { + return nil, "", err + } + if opt.UpTo != "" { + chain, err = truncateChain(chain, opt.UpTo) + if err != nil { + return nil, "", err + } + } + return d.Runner.Run(parent, jobs.Job{ Stage: "restore", Func: func(ctx context.Context, emit func(jobs.Event)) error { @@ -38,20 +54,59 @@ func Restore(parent context.Context, d Deps, opt RestoreOpts) (<-chan jobs.Event } defer func() { _ = conn.Close() }() - f, err := os.Open(d.Dumps.Path(opt.DumpID)) - if err != nil { - return err + for i, m := range chain { + emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "applying " + m.ID}) + f, err := os.Open(d.Dumps.Path(m.ID)) + if err != nil { + return err + } + env, body, err := dumps.ReadEnvelope(f) + if err != nil { + _ = f.Close() + return err + } + // Guard against a destructive Clean wiping the target before we + // discover the dump was produced by a different engine. Verify + // EVERY dump in the chain matches the target driver before the + // first (Clean) restore can run. + if env.Driver != resolved.Driver { + _ = f.Close() + return &errs.Error{ + Op: "restore", + Code: errs.CodeUser, + Cause: errs.ErrIncompatibleEngine, + Hint: "dump was created by " + env.Driver + "; cannot restore into a " + resolved.Driver + " target", + } + } + rOpts := driver.RestoreOpts{ + TargetTables: opt.TargetTables, + SchemaOnly: opt.SchemaOnly, + DataOnly: opt.DataOnly, + Clean: opt.Clean && i == 0, // clean once, before the base only + } + if err := conn.Restore(ctx, rOpts, body); err != nil { + _ = f.Close() + return err + } + _ = f.Close() } - defer func() { _ = f.Close() }() - - emit(jobs.Event{Phase: jobs.PhaseProgress, Message: "restoring " + opt.DumpID}) - - return conn.Restore(ctx, driver.RestoreOpts{ - TargetTables: opt.TargetTables, - SchemaOnly: opt.SchemaOnly, - DataOnly: opt.DataOnly, - Clean: opt.Clean, - }, f) + return nil }, }) } + +// truncateChain returns chain up to and including the dump named upTo. Unlike +// silently applying the full chain, an unknown upTo is an error so a typo'd +// --up-to surfaces instead of restoring more than the user asked for. +func truncateChain(chain []dumps.Meta, upTo string) ([]dumps.Meta, error) { + for i, m := range chain { + if m.ID == upTo { + return chain[:i+1], nil + } + } + return nil, &errs.Error{ + Op: "restore", + Code: errs.CodeUser, + Hint: "--up-to " + upTo + " is not in the restore chain for the target dump", + } +} diff --git a/internal/app/restore_chain_test.go b/internal/app/restore_chain_test.go new file mode 100644 index 0000000..1dd43d7 --- /dev/null +++ b/internal/app/restore_chain_test.go @@ -0,0 +1,324 @@ +package app + +import ( + "context" + "errors" + "io" + "os" + "testing" + "time" + + "github.com/nixrajput/siphon/internal/config" + "github.com/nixrajput/siphon/internal/driver" + "github.com/nixrajput/siphon/internal/dumps" + "github.com/nixrajput/siphon/internal/errs" + "github.com/nixrajput/siphon/internal/jobs" + "github.com/nixrajput/siphon/internal/profile" + "github.com/nixrajput/siphon/internal/secrets" +) + +// chainRestoreCall records one Restore invocation: the native body the driver +// received (envelope already stripped) and whether Clean was requested. +type chainRestoreCall struct { + body string + clean bool +} + +// chainFakeConn records every Restore call so a test can assert chain order, +// per-dump envelope stripping, and that Clean was set only for the base. +type chainFakeConn struct { + calls []chainRestoreCall +} + +func (c *chainFakeConn) Inspect(_ context.Context) (*driver.Schema, error) { + return &driver.Schema{}, nil +} + +func (c *chainFakeConn) Backup(_ context.Context, _ driver.BackupOpts, _ io.Writer) error { + return nil +} + +func (c *chainFakeConn) Restore(_ context.Context, opts driver.RestoreOpts, r io.Reader) error { + b, err := io.ReadAll(r) + if err != nil { + return err + } + c.calls = append(c.calls, chainRestoreCall{body: string(b), clean: opts.Clean}) + return nil +} + +func (c *chainFakeConn) Verify(_ context.Context, _ io.Reader) (*driver.VerifyReport, error) { + return &driver.VerifyReport{OK: true}, nil +} + +func (c *chainFakeConn) Close() error { return nil } + +// chainFakeDriver always returns the same chainFakeConn so the test inspects it. +type chainFakeDriver struct{ conn driver.Conn } + +func (d *chainFakeDriver) Name() string { return "fake" } +func (d *chainFakeDriver) Capabilities() driver.Capabilities { + return driver.Capabilities{NativeStream: true} +} +func (d *chainFakeDriver) Connect(_ context.Context, _ driver.Profile) (driver.Conn, error) { + return d.conn, nil +} + +// writeChainDump writes a real catalog dump file (envelope header + native +// payload) at cat.Path(id) plus its sidecar Meta with the given lineage. +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) + } + typ := dumps.EnvelopeIncremental + if parentID == "" { + typ = dumps.EnvelopeBase + } + if _, err := dumps.WriteEnvelope(f, &dumps.Envelope{ + Type: typ, + Driver: "fake", + BaseID: baseID, + ParentID: parentID, + }); 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) + } + if err := cat.WriteMeta(&dumps.Meta{ + ID: id, + Profile: "test", + Driver: "fake", + BaseID: baseID, + ParentID: parentID, + }); err != nil { + t.Fatalf("write meta %s: %v", id, err) + } +} + +// chainDeps builds a Deps backed by a real catalog at dir and the given +// recording conn, with a single profile. +func chainDeps(t *testing.T, dir string, conn driver.Conn) Deps { + t.Helper() + t.Setenv("SIPHON_CONFIG_HOME", t.TempDir()) + cfg := &config.Config{Profiles: map[string]config.ProfileConfig{ + "test": {Driver: "fake", Host: "h", User: "u", Database: "d", Password: "p"}, + }} + res := secrets.NewResolver(secrets.Passthrough{}) + ps := profile.New(cfg, res, func(*config.Config) error { return nil }) + cat, err := dumps.NewCatalog(dir) + if err != nil { + t.Fatalf("NewCatalog: %v", err) + } + return Deps{ + Profiles: ps, + Dumps: cat, + Runner: jobs.NewRunner(), + Drivers: fakeGetter{d: &chainFakeDriver{conn: conn}}, + } +} + +// seedThreeDumpChain writes base -> inc1 -> inc2 into a fresh catalog at dir and +// returns the Deps wired to conn. +func seedThreeDumpChain(t *testing.T, dir string, conn driver.Conn) Deps { + t.Helper() + deps := chainDeps(t, dir, conn) + cat := deps.Dumps + writeChainDump(t, cat, "base", "base", "", "base-data") + writeChainDump(t, cat, "inc1", "base", "base", "inc1-data") + writeChainDump(t, cat, "inc2", "base", "inc1", "inc2-data") + return deps +} + +// TestRestoreChain_AppliesInOrder proves the chain is applied base->inc1->inc2 +// and that each recorded body is the native payload with the 4 KB envelope +// stripped per dump (not the raw file bytes). +func TestRestoreChain_AppliesInOrder(t *testing.T) { + conn := &chainFakeConn{} + deps := seedThreeDumpChain(t, t.TempDir(), conn) + + ch, _, err := Restore(context.Background(), deps, RestoreOpts{Profile: "test", DumpID: "inc2"}) + if err != nil { + t.Fatalf("Restore: %v", err) + } + drain(t, ch) + + want := []string{"base-data", "inc1-data", "inc2-data"} + if len(conn.calls) != len(want) { + t.Fatalf("Restore made %d calls; want %d", len(conn.calls), len(want)) + } + for i, w := range want { + if conn.calls[i].body != w { + t.Fatalf("call %d body = %q; want %q (envelope must be stripped per dump)", i, conn.calls[i].body, w) + } + } +} + +// TestRestoreChain_CleanOnlyBeforeBase proves Clean is set on the base restore +// only and false for every incremental, when Restore is called with Clean:true. +func TestRestoreChain_CleanOnlyBeforeBase(t *testing.T) { + conn := &chainFakeConn{} + deps := seedThreeDumpChain(t, t.TempDir(), conn) + + ch, _, err := Restore(context.Background(), deps, RestoreOpts{Profile: "test", DumpID: "inc2", Clean: true}) + if err != nil { + t.Fatalf("Restore: %v", err) + } + drain(t, ch) + + if len(conn.calls) != 3 { + t.Fatalf("Restore made %d calls; want 3", len(conn.calls)) + } + if !conn.calls[0].clean { + t.Fatalf("base call clean = false; want true") + } + for i := 1; i < len(conn.calls); i++ { + if conn.calls[i].clean { + t.Fatalf("incremental call %d clean = true; want false (clean once, base only)", i) + } + } +} + +// TestRestoreChain_UpToTruncates proves --up-to stops the chain early, applying +// only base and inc1 when targeting inc2 with UpTo=inc1. +func TestRestoreChain_UpToTruncates(t *testing.T) { + conn := &chainFakeConn{} + deps := seedThreeDumpChain(t, t.TempDir(), conn) + + ch, _, err := Restore(context.Background(), deps, RestoreOpts{Profile: "test", DumpID: "inc2", UpTo: "inc1"}) + if err != nil { + t.Fatalf("Restore: %v", err) + } + drain(t, ch) + + want := []string{"base-data", "inc1-data"} + if len(conn.calls) != len(want) { + t.Fatalf("Restore made %d calls; want %d", len(conn.calls), len(want)) + } + for i, w := range want { + if conn.calls[i].body != w { + t.Fatalf("call %d body = %q; want %q", i, conn.calls[i].body, w) + } + } +} + +// TestRestoreChain_UpToUnknownErrors proves an --up-to that names a dump not in +// the chain is a synchronous CodeUser error (returned before the job runs), not +// a silent full-chain restore. +func TestRestoreChain_UpToUnknownErrors(t *testing.T) { + conn := &chainFakeConn{} + deps := seedThreeDumpChain(t, t.TempDir(), conn) + + _, _, err := Restore(context.Background(), deps, RestoreOpts{Profile: "test", DumpID: "inc2", UpTo: "nonexistent"}) + if err == nil { + t.Fatal("Restore with unknown --up-to returned nil error; want CodeUser error") + } + var e *errs.Error + if !errors.As(err, &e) { + t.Fatalf("error type = %T; want *errs.Error", err) + } + if e.Code != errs.CodeUser { + t.Fatalf("error Code = %d; want CodeUser (%d)", e.Code, errs.CodeUser) + } + if len(conn.calls) != 0 { + t.Fatalf("driver received %d Restore calls; want 0 (error must be synchronous)", len(conn.calls)) + } +} + +// writeMismatchDump writes a single base dump whose envelope.Driver differs +// from the target profile's driver ("fake"), so Restore must reject it before +// 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{ + 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) + } + if err := cat.WriteMeta(&dumps.Meta{ + ID: id, + Profile: "test", + Driver: envDriver, + BaseID: id, + }); err != nil { + t.Fatalf("write meta %s: %v", id, err) + } +} + +// TestRestoreChain_DriverMismatchRejectsBeforeClean proves a dump whose +// envelope.Driver differs from the target profile's driver makes Restore return +// a CodeUser/ErrIncompatibleEngine error and NEVER calls the driver's Restore — +// so a destructive Clean cannot wipe the target before the mismatch is found. +func TestRestoreChain_DriverMismatchRejectsBeforeClean(t *testing.T) { + conn := &chainFakeConn{} + dir := t.TempDir() + deps := chainDeps(t, dir, conn) + // Target profile driver is "fake"; this dump claims "postgres". + writeMismatchDump(t, deps.Dumps, "base", "postgres") + + ch, _, err := Restore(context.Background(), deps, RestoreOpts{Profile: "test", DumpID: "base", Clean: true}) + if err != nil { + t.Fatalf("Restore setup: %v", err) + } + gotErr := drainErr(t, ch) + if gotErr == nil { + t.Fatal("Restore with mismatched driver returned nil error; want incompatible-engine error") + } + var e *errs.Error + if !errors.As(gotErr, &e) { + t.Fatalf("error type = %T; want *errs.Error", gotErr) + } + if e.Code != errs.CodeUser { + t.Fatalf("error Code = %d; want CodeUser (%d)", e.Code, errs.CodeUser) + } + if !errors.Is(gotErr, errs.ErrIncompatibleEngine) { + t.Fatalf("error = %v; want ErrIncompatibleEngine", gotErr) + } + if len(conn.calls) != 0 { + t.Fatalf("driver Restore called %d times; want 0 (no destructive Clean on mismatch)", len(conn.calls)) + } +} + +// drainErr consumes a job event channel and returns the error from a +// PhaseError event (or nil if the job completed cleanly), with a hard timeout. +func drainErr(t *testing.T, ch <-chan jobs.Event) error { + t.Helper() + timer := time.NewTimer(5 * time.Second) + defer timer.Stop() + var jobErr error + for { + select { + case ev, ok := <-ch: + if !ok { + return jobErr + } + if ev.Phase == jobs.PhaseError && ev.Err != nil { + jobErr = ev.Err + } + case <-timer.C: + t.Fatal("job did not complete within 5 s") + return nil + } + } +} diff --git a/internal/app/sync.go b/internal/app/sync.go index 2792aea..ae64e8b 100644 --- a/internal/app/sync.go +++ b/internal/app/sync.go @@ -2,9 +2,9 @@ package app import ( "context" - "io" "github.com/nixrajput/siphon/internal/driver" + "github.com/nixrajput/siphon/internal/errs" "github.com/nixrajput/siphon/internal/jobs" ) @@ -14,13 +14,33 @@ type SyncOpts struct { To string Stream bool Tables []string + // CrossEngine routes through the canonical-schema path (e.g. postgres→mysql) + // instead of the native homogeneous stream. Gated on driver capability. + CrossEngine bool + // Continuous requests CDC follow mode. Not yet available (Phase F ships CDC + // as an internal scaffold only). Setting it returns a clear CodeUser error. + Continuous bool } -// Sync backs up From and restores into To. In Phase B the data always flows -// through an in-memory io.Pipe (no temp file, no catalog entry). The opt.Stream -// flag and Capabilities().NativeStream gating — and a catalog fallback for -// drivers that can't stream — are planned for Phase F; today Stream is unused. +// Sync backs up From and restores into To in a single pass. The native +// (homogeneous) path streams the dump through a bounded jobs.Stream — no temp +// file, no catalog entry — so backpressure is observable via FillPercent while +// a backup failure still propagates to Restore as a read error (via CloseErr), +// preventing a truncated dump from being committed as if clean. +// +// When opt.CrossEngine is set the work routes through runCrossEngineSync, which +// is capability-gated and today returns a clear unsupported error (no driver +// declares cross-engine source/target yet). func Sync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, string, error) { + if opt.Continuous { + return nil, "", &errs.Error{ + Op: "sync.continuous", + Code: errs.CodeUser, + Cause: errs.ErrDriverUnsupported, + Hint: "continuous CDC sync is not yet available (Phase F follow-up); see docs/CDC.md", + } + } + src, err := d.Profiles.Resolve(opt.From) if err != nil { return nil, "", err @@ -38,6 +58,10 @@ func Sync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, stri return nil, "", err } + if opt.CrossEngine { + return runCrossEngineSync(parent, d, opt) + } + return d.Runner.Run(parent, jobs.Job{ Stage: "sync", Func: func(ctx context.Context, emit func(jobs.Event)) error { @@ -53,21 +77,25 @@ func Sync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, stri } defer func() { _ = dstConn.Close() }() - pr, pw := io.Pipe() + // Stream is both the backup writer and the restore reader (it is + // io.Reader+io.Writer): the backup goroutine writes, Restore reads. + stream := jobs.NewStream(64) errCh := make(chan error, 1) go func() { - bErr := srcConn.Backup(ctx, driver.BackupOpts{IncludeTables: opt.Tables}, pw) - // CloseWithError propagates a backup failure to the reader as a - // read error instead of a clean io.EOF. Without this, a truncated - // dump looks complete to Restore — which, with Clean:true, has - // already dropped the target and would commit partial data. - _ = pw.CloseWithError(bErr) + bErr := srcConn.Backup(ctx, driver.BackupOpts{IncludeTables: opt.Tables}, stream) + // CloseErr propagates a backup failure to the reader as a read + // error instead of a clean io.EOF (the bounded-buffer analogue of + // io.PipeWriter.CloseWithError). Without this, a truncated dump + // looks complete to Restore — which, with Clean:true, has already + // dropped the target and would commit partial data. A nil bErr + // behaves like Close → clean EOF. + _ = stream.CloseErr(bErr) errCh <- bErr }() - restoreErr := dstConn.Restore(ctx, driver.RestoreOpts{TargetTables: opt.Tables, Clean: true}, pr) - _ = pr.Close() // unblock the backup goroutine's pw.Write immediately if Restore returned early + restoreErr := dstConn.Restore(ctx, driver.RestoreOpts{TargetTables: opt.Tables, Clean: true}, stream) + _ = stream.Close() // unblock the backup goroutine's Write immediately if Restore returned early backupErr := <-errCh if backupErr != nil { @@ -77,3 +105,35 @@ func Sync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, stri }, }) } + +// runCrossEngineSync handles heterogeneous sync (e.g. postgres→mysql) by routing +// through the canonical-schema machinery (driver EmitCanonical/ConsumeCanonical). +// +// It is HONESTLY gated: it checks CapCrossEngineSource on the source driver and +// CapCrossEngineTarget on the target. No driver declares either today (all +// return false — see each driver.go), because cross-engine translation needs a +// typed CanonicalSchema and driver.Inspect carries no column types yet. So the +// capability gate rejects every cross-engine request with ErrDriverUnsupported. +// +// This is deliberate scaffolding: the canonical emit/consume machinery (Task 8) +// exists and is unit-tested, but wiring it requires typed schema introspection +// (a future task). We do NOT fabricate a CanonicalSchema here. Once Inspect +// emits column types and a driver flips its cross-engine caps to true, this +// function gains the real emit→translate→consume pipeline. +func runCrossEngineSync(parent context.Context, d Deps, opt SyncOpts) (<-chan jobs.Event, string, error) { + if err := RequireCapability(d, opt.From, CapCrossEngineSource); err != nil { + return nil, "", err + } + if err := RequireCapability(d, opt.To, CapCrossEngineTarget); err != nil { + return nil, "", err + } + // Unreachable today (no driver advertises cross-engine caps), but kept as a + // clear, honest backstop should a cap ever flip true before the pipeline is + // actually wired. + return nil, "", &errs.Error{ + Op: "sync.cross_engine", + Code: errs.CodeUser, + Cause: errs.ErrDriverUnsupported, + Hint: "cross-engine sync requires typed schema introspection, not yet available", + } +} diff --git a/internal/app/sync_test.go b/internal/app/sync_test.go index c09e849..9a702fb 100644 --- a/internal/app/sync_test.go +++ b/internal/app/sync_test.go @@ -1,7 +1,6 @@ package app import ( - "bytes" "context" "errors" "io" @@ -17,8 +16,9 @@ import ( ) // syncFakeConn is a fakeConn variant whose Restore behaviour is configurable. -// Backup writes a payload large enough (>64 KiB) to fill the kernel pipe buffer -// so that pw.Write would block indefinitely if pr is never closed. +// Backup writes a payload large enough to overflow the bounded jobs.Stream +// buffer so that Write blocks indefinitely if the stream is never closed after +// Restore returns early. type syncFakeConn struct { restoreErr error backupData []byte @@ -29,8 +29,23 @@ func (c *syncFakeConn) Inspect(_ context.Context) (*driver.Schema, error) { } func (c *syncFakeConn) Backup(_ context.Context, _ driver.BackupOpts, w io.Writer) error { - _, err := io.Copy(w, bytes.NewReader(c.backupData)) - return err + // Write in many small chunks rather than io.Copy(w, bytes.NewReader(...)): + // bytes.Reader implements io.WriterTo, so io.Copy would collapse the whole + // payload into a SINGLE w.Write call, which a bounded jobs.Stream accepts as + // one buffered chunk and never blocks on. Looping forces ≥64 separate Writes + // so the bounded buffer actually fills and the producer blocks when the + // reader (Restore) stops draining — which is the leak scenario under test. + const chunk = 64 * 1024 + for off := 0; off < len(c.backupData); off += chunk { + end := off + chunk + if end > len(c.backupData) { + end = len(c.backupData) + } + if _, err := w.Write(c.backupData[off:end]); err != nil { + return err + } + } + return nil } func (c *syncFakeConn) Restore(_ context.Context, _ driver.RestoreOpts, _ io.Reader) error { @@ -75,9 +90,13 @@ func (g syncDualGetter) Get(name string) (driver.Driver, error) { // channel closes) within a short timeout. Before the pr.Close() fix the backup // goroutine's pw.Write would block forever, causing Sync to hang. func TestSync_PipeReaderClosed_NoGoroutineLeak(t *testing.T) { - // 128 KiB > typical pipe buffer (64 KiB on Linux/macOS), so pw.Write - // blocks if pr is not closed after Restore returns early. - bigPayload := make([]byte, 128*1024) + // The native sync path streams through a bounded jobs.Stream (64 chunks). + // io.Copy feeds it in ~32 KiB chunks, so the buffer holds ~2 MiB before a + // Write blocks. Use 8 MiB so the producer is guaranteed to block once the + // 64-chunk buffer fills — then if Restore returns early and the stream is + // NOT closed, the backup goroutine stays parked on Write forever (leak). + // Sync must close the stream after Restore returns to unblock it. + bigPayload := make([]byte, 8*1024*1024) restoreErr := errors.New("restore failed intentionally") diff --git a/internal/cli/backup.go b/internal/cli/backup.go index 735b310..6daf6e1 100644 --- a/internal/cli/backup.go +++ b/internal/cli/backup.go @@ -4,6 +4,7 @@ import ( "github.com/spf13/cobra" "github.com/nixrajput/siphon/internal/app" + "github.com/nixrajput/siphon/internal/errs" ) func newBackupCmd() *cobra.Command { @@ -16,6 +17,8 @@ func newBackupCmd() *cobra.Command { dataOnly bool parallel int compressionLvl int + incremental bool + baseID string ) cmd := &cobra.Command{ Use: "backup [profile]", @@ -25,6 +28,31 @@ func newBackupCmd() *cobra.Command { if len(args) == 1 { profileName = args[0] } + // Incremental backup is scaffolded but not wired end-to-end. Full + // wiring needs: (a) reading --base's envelope for the parent + // WAL/binlog position, (b) an optional-interface dance to reach the + // driver's BackupIncremental (it is on the concrete *Conn types, not + // driver.Conn), and (c) writing an incremental-type envelope with + // BaseID/ParentID. It also needs a live DB to verify, plus the + // tracked runtime gates: orphan replication-slot cleanup for + // Postgres and binlog-position validation for MySQL/MariaDB. Tracked + // as a Phase F follow-up; rejected honestly here rather than + // half-wired into app.Backup. + if baseID != "" && !incremental { + return &errs.Error{ + Op: "backup", + Code: errs.CodeUser, + Hint: "--base requires --incremental", + } + } + if incremental { + return &errs.Error{ + Op: "backup.incremental", + Code: errs.CodeUser, + Cause: errs.ErrDriverUnsupported, + Hint: "incremental backup is not yet wired end-to-end (Phase F follow-up); the driver-level machinery exists", + } + } deps, err := buildDeps() if err != nil { return err @@ -53,5 +81,7 @@ func newBackupCmd() *cobra.Command { cmd.Flags().BoolVar(&dataOnly, "data-only", false, "Data, no schema") cmd.Flags().IntVar(¶llel, "jobs", 1, "Parallel workers (not yet effective for backup; Phase F)") cmd.Flags().IntVar(&compressionLvl, "compression", 1, "Compression level 0-9") + cmd.Flags().BoolVar(&incremental, "incremental", false, "Capture only changes since --base (not yet wired end-to-end; Phase F follow-up)") + cmd.Flags().StringVar(&baseID, "base", "", "Base dump ID for an incremental backup (used with --incremental)") return cmd } diff --git a/internal/cli/restore.go b/internal/cli/restore.go index 7a117e0..d51b328 100644 --- a/internal/cli/restore.go +++ b/internal/cli/restore.go @@ -16,6 +16,7 @@ func newRestoreCmd() *cobra.Command { schemaOnly bool dataOnly bool clean bool + upTo string ) cmd := &cobra.Command{ Use: "restore [dump-id]", @@ -39,6 +40,7 @@ func newRestoreCmd() *cobra.Command { SchemaOnly: schemaOnly, DataOnly: dataOnly, Clean: clean, + UpTo: upTo, }) if err != nil { return err @@ -52,5 +54,6 @@ func newRestoreCmd() *cobra.Command { cmd.Flags().BoolVar(&schemaOnly, "schema-only", false, "Schema, no data") cmd.Flags().BoolVar(&dataOnly, "data-only", false, "Data, no schema") cmd.Flags().BoolVar(&clean, "clean", false, "DROP objects before recreating") + cmd.Flags().StringVar(&upTo, "up-to", "", "Stop applying the incremental chain after this dump ID") return cmd } diff --git a/internal/cli/sync.go b/internal/cli/sync.go index 2518b5f..0849eb7 100644 --- a/internal/cli/sync.go +++ b/internal/cli/sync.go @@ -11,6 +11,8 @@ func newSyncCmd() *cobra.Command { fromName, toName string stream bool tables []string + crossEngine bool + continuous bool ) cmd := &cobra.Command{ Use: "sync [from] [to]", @@ -29,6 +31,7 @@ func newSyncCmd() *cobra.Command { } ch, _, err := app.Sync(c.Context(), deps, app.SyncOpts{ From: fromName, To: toName, Stream: stream, Tables: tables, + CrossEngine: crossEngine, Continuous: continuous, }) if err != nil { return err @@ -40,5 +43,7 @@ func newSyncCmd() *cobra.Command { cmd.Flags().StringVar(&toName, "to", "", "Target profile") cmd.Flags().BoolVar(&stream, "stream", true, "Stream source→target without intermediate file") cmd.Flags().StringSliceVar(&tables, "table", nil, "Limit to these tables") + cmd.Flags().BoolVar(&crossEngine, "cross-engine", false, "Translate between engines via canonical schema (requires cross-engine driver support; not yet available)") + cmd.Flags().BoolVar(&continuous, "continuous", false, "Continuously follow source changes (CDC) — not yet available; see docs/CDC.md") return cmd } diff --git a/internal/driver/_mysqlcommon/conn.go b/internal/driver/_mysqlcommon/conn.go index 615f68d..a291521 100644 --- a/internal/driver/_mysqlcommon/conn.go +++ b/internal/driver/_mysqlcommon/conn.go @@ -33,6 +33,7 @@ type Conn struct { p driver.Profile dumpBinary string // "mysqldump" or "mariadb-dump" clientBinary string // "mysql" or "mariadb" + binlogBinary string // "mysqlbinlog" or "mariadb-binlog" } var _ driver.Conn = (*Conn)(nil) @@ -40,8 +41,9 @@ var _ driver.Conn = (*Conn)(nil) // NewConn opens + pings the database and returns a ready Conn. The ping is // wrapped in a bounded retry (jobs.Retry, 3 attempts) — same policy as the // Postgres driver (spec §4.3). connOp is the error-wrapping op label, e.g. -// "mysql.connect" / "mariadb.connect", so errors name the right driver. -func NewConn(ctx context.Context, p driver.Profile, dumpBinary, clientBinary, connOp string) (*Conn, error) { +// "mysql.connect" / "mariadb.connect", so errors name the right driver. The +// three binary names (dump/client/binlog) are the only per-fork difference. +func NewConn(ctx context.Context, p driver.Profile, dumpBinary, clientBinary, binlogBinary, connOp string) (*Conn, error) { db, err := Open(p) if err != nil { return nil, connErr(connOp, err) @@ -50,7 +52,7 @@ func NewConn(ctx context.Context, p driver.Profile, dumpBinary, clientBinary, co _ = db.Close() return nil, connErr(connOp, err) } - return &Conn{db: db, p: p, dumpBinary: dumpBinary, clientBinary: clientBinary}, nil + return &Conn{db: db, p: p, dumpBinary: dumpBinary, clientBinary: clientBinary, binlogBinary: binlogBinary}, nil } // connErr wraps a connection failure in the shape the harness asserts on diff --git a/internal/driver/_mysqlcommon/incremental.go b/internal/driver/_mysqlcommon/incremental.go new file mode 100644 index 0000000..02bc509 --- /dev/null +++ b/internal/driver/_mysqlcommon/incremental.go @@ -0,0 +1,170 @@ +package mysqlcommon + +import ( + "context" + "database/sql" + "errors" + "io" + "os" + "os/exec" + "strconv" + "strings" + + "github.com/nixrajput/siphon/internal/driver" + "github.com/nixrajput/siphon/internal/errs" +) + +// BinlogPosition identifies the binlog file + offset where incremental events +// begin. Captured at base-backup time and stored in the dump Envelope. +type BinlogPosition struct { + File string // e.g. "mysql-bin.000123" + Position uint64 +} + +// CaptureBinlogPosition records the current binlog coordinates. Tries the +// MySQL 8.4+ statement first, then the pre-8.4 form, so it works across +// versions and both forks. +func (c *Conn) CaptureBinlogPosition(ctx context.Context) (BinlogPosition, error) { + // SHOW BINARY LOG STATUS (MySQL 8.4+) and SHOW MASTER STATUS (older MySQL / + // MariaDB) return the same first two columns (File, Position) but differ in + // trailing columns, so scan into a flexible set. We read File+Position via a + // columns-agnostic approach: fetch all columns, take the first two. + for _, q := range []string{"SHOW BINARY LOG STATUS", "SHOW MASTER STATUS"} { + pos, ok := tryBinlogStatus(ctx, c.db, q) + if ok { + return pos, nil + } + } + return BinlogPosition{}, &errs.Error{ + Op: "mysql.binlog_position", + Code: errs.CodeSystem, + Cause: errors.New("could not read binlog position"), + Hint: "ensure log_bin = ON and binlog_format = ROW on the source", + } +} + +// tryBinlogStatus runs a SHOW ... STATUS query and extracts File+Position from +// the first two columns, tolerating the differing trailing-column counts across +// MySQL versions and MariaDB by scanning into []sql.RawBytes sized to the +// actual column count. +func tryBinlogStatus(ctx context.Context, db *sql.DB, query string) (BinlogPosition, bool) { + rows, err := db.QueryContext(ctx, query) + if err != nil { + return BinlogPosition{}, false + } + defer func() { _ = rows.Close() }() + cols, err := rows.Columns() + if err != nil || len(cols) < 2 { + return BinlogPosition{}, false + } + if !rows.Next() { + return BinlogPosition{}, false + } + raw := make([]sql.RawBytes, len(cols)) + scanArgs := make([]any, len(cols)) + for i := range raw { + scanArgs[i] = &raw[i] + } + if err := rows.Scan(scanArgs...); err != nil { + return BinlogPosition{}, false + } + file := string(raw[0]) + pos, err := strconv.ParseUint(string(raw[1]), 10, 64) + if err != nil || file == "" { + return BinlogPosition{}, false + } + return BinlogPosition{File: file, Position: pos}, true +} + +// ValidateBinlogFormat returns a CodeUser error if binlog_format != ROW. +func (c *Conn) ValidateBinlogFormat(ctx context.Context) error { + var format string + if err := c.db.QueryRowContext(ctx, "SELECT @@binlog_format").Scan(&format); err != nil { + return &errs.Error{Op: "mysql.incremental", Code: errs.CodeSystem, Cause: err} + } + if !strings.EqualFold(format, "ROW") { + return &errs.Error{ + Op: "mysql.incremental", + Code: errs.CodeUser, + Cause: errors.New("binlog_format is " + format), + Hint: "set binlog_format = ROW on the source and restart the server", + } + } + return nil +} + +// binlogArgs builds the argument vector for the fork's binlog tool +// (mysqlbinlog / mariadb-binlog). The password is passed via MYSQL_PWD in the +// environment (see BackupIncremental), never on the command line. The starting +// binlog file is the final positional arg; --to-last-log continues through all +// subsequent rotated files to the current end of the binlog. +// +// SSL flags are fork-specific, keyed on binlogBinary: mysqlbinlog takes +// --ssl-mode=, while +// mariadb-binlog takes --ssl / --skip-ssl. The starting binlog file MUST stay +// last (positional), so SSL flags are inserted before it. +func binlogArgs(p driver.Profile, since BinlogPosition, binlogBinary string) []string { + args := []string{ + "-h", p.Host, + "-P", strconv.Itoa(p.Port), + "-u", p.User, + "--read-from-remote-server", + "--to-last-log", + "--start-position=" + strconv.FormatUint(since.Position, 10), + } + args = append(args, binlogSSLArgs(p.SSLMode, binlogBinary)...) + return append(args, since.File) +} + +// binlogSSLArgs maps Profile.SSLMode to the fork-specific binlog TLS flags. +// mysqlbinlog uses --ssl-mode=; mariadb-binlog uses --ssl / --skip-ssl. +// An empty/PREFERRED policy omits the flag entirely (the tool's default). +func binlogSSLArgs(sslMode, binlogBinary string) []string { + switch binlogBinary { + case "mysqlbinlog": + var level string + switch sslMode { + case "disable": + level = "DISABLED" + case "require": + level = "REQUIRED" + case "verify-ca": + level = "VERIFY_CA" + case "verify-full": + level = "VERIFY_IDENTITY" + default: + level = "PREFERRED" + } + return []string{"--ssl-mode=" + level} + case "mariadb-binlog": + switch sslMode { + case "require", "verify-ca", "verify-full": + return []string{"--ssl"} + case "disable": + return []string{"--skip-ssl"} + default: + return nil // PREFERRED/empty: omit, use the tool's default + } + default: + return nil + } +} + +// BackupIncremental streams binlog events from `since` to current end-of-binlog +// into w, using the fork's binlog tool (mysqlbinlog / mariadb-binlog). +// +// NOTE: the --read-from-remote-server + --start-position invocation is +// structurally complete but UNPROVEN locally (no Docker/MySQL here). The exact +// remote-auth flags and whether a single starting binlog file suffices vs. +// needing --to-last-log need validation against a live log_bin=ON, +// binlog_format=ROW server (see CI / the incremental wiring task). +func (c *Conn) BackupIncremental(ctx context.Context, since BinlogPosition, w io.Writer) error { + cmd := exec.CommandContext(ctx, c.binlogBinary, binlogArgs(c.p, since, c.binlogBinary)...) + cmd.Env = withMySQLPwd(os.Environ(), c.p.Password) + cmd.Stdout = w + cmd.Stderr = os.Stderr + if err := cmd.Run(); err != nil { + return &errs.Error{Op: c.binlogBinary + ".backup_incremental", Code: errs.CodeSystem, Cause: err} + } + return nil +} diff --git a/internal/driver/_mysqlcommon/incremental_test.go b/internal/driver/_mysqlcommon/incremental_test.go new file mode 100644 index 0000000..aaddb81 --- /dev/null +++ b/internal/driver/_mysqlcommon/incremental_test.go @@ -0,0 +1,58 @@ +package mysqlcommon + +import ( + "reflect" + "slices" + "testing" + + "github.com/nixrajput/siphon/internal/driver" +) + +func TestBinlogArgs(t *testing.T) { + p := driver.Profile{Host: "db.local", Port: 3306, User: "root", Database: "shop"} + since := BinlogPosition{File: "mysql-bin.000123", Position: 4096} + got := binlogArgs(p, since, "mysqlbinlog") + want := []string{ + "-h", "db.local", + "-P", "3306", + "-u", "root", + "--read-from-remote-server", + "--to-last-log", + "--start-position=4096", + "--ssl-mode=PREFERRED", // no SSLMode set => PREFERRED for mysqlbinlog + "mysql-bin.000123", + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("binlogArgs() = %v\nwant %v", got, want) + } + // The starting binlog file must remain the final positional arg. + if got[len(got)-1] != "mysql-bin.000123" { + t.Fatalf("binlog file not last arg: %v", got) + } +} + +// TestBinlogArgs_SSL asserts the fork-specific TLS flags: mysqlbinlog maps +// SSLMode to --ssl-mode=, while mariadb-binlog uses --ssl / --skip-ssl. +func TestBinlogArgs_SSL(t *testing.T) { + p := driver.Profile{Host: "h", Port: 3306, User: "u", SSLMode: "require"} + since := BinlogPosition{File: "bin.1", Position: 0} + + mysql := binlogArgs(p, since, "mysqlbinlog") + if !slices.Contains(mysql, "--ssl-mode=REQUIRED") { + t.Fatalf("mysqlbinlog with require: missing --ssl-mode=REQUIRED; got %v", mysql) + } + + maria := binlogArgs(p, since, "mariadb-binlog") + if !slices.Contains(maria, "--ssl") { + t.Fatalf("mariadb-binlog with require: missing --ssl; got %v", maria) + } + + // disable maps to --skip-ssl for mariadb-binlog and DISABLED for mysqlbinlog. + pDis := driver.Profile{Host: "h", Port: 3306, User: "u", SSLMode: "disable"} + if got := binlogArgs(pDis, since, "mariadb-binlog"); !slices.Contains(got, "--skip-ssl") { + t.Fatalf("mariadb-binlog with disable: missing --skip-ssl; got %v", got) + } + if got := binlogArgs(pDis, since, "mysqlbinlog"); !slices.Contains(got, "--ssl-mode=DISABLED") { + t.Fatalf("mysqlbinlog with disable: missing --ssl-mode=DISABLED; got %v", got) + } +} diff --git a/internal/driver/mariadb/driver.go b/internal/driver/mariadb/driver.go index 6f80a10..0ae124d 100644 --- a/internal/driver/mariadb/driver.go +++ b/internal/driver/mariadb/driver.go @@ -18,7 +18,7 @@ func (Driver) Name() string { return "mariadb" } func (Driver) Capabilities() driver.Capabilities { return driver.Capabilities{ - Incremental: false, // Phase F (binlog) + Incremental: true, // binlog (mysqlbinlog/mariadb-binlog) NativeStream: true, PerTable: true, SchemaOnly: true, @@ -35,7 +35,7 @@ func (Driver) Capabilities() driver.Capabilities { } func (Driver) Connect(ctx context.Context, p driver.Profile) (driver.Conn, error) { - return mysqlcommon.NewConn(ctx, p, "mariadb-dump", "mariadb", "mariadb.connect") + return mysqlcommon.NewConn(ctx, p, "mariadb-dump", "mariadb", "mariadb-binlog", "mariadb.connect") } // Compile-time check. diff --git a/internal/driver/mysql/driver.go b/internal/driver/mysql/driver.go index 68e6e6f..25ef55f 100644 --- a/internal/driver/mysql/driver.go +++ b/internal/driver/mysql/driver.go @@ -18,7 +18,7 @@ func (Driver) Name() string { return "mysql" } func (Driver) Capabilities() driver.Capabilities { return driver.Capabilities{ - Incremental: false, // Phase F (binlog) + Incremental: true, // binlog (mysqlbinlog/mariadb-binlog) NativeStream: true, PerTable: true, SchemaOnly: true, @@ -35,7 +35,7 @@ func (Driver) Capabilities() driver.Capabilities { } func (Driver) Connect(ctx context.Context, p driver.Profile) (driver.Conn, error) { - return mysqlcommon.NewConn(ctx, p, "mysqldump", "mysql", "mysql.connect") + return mysqlcommon.NewConn(ctx, p, "mysqldump", "mysql", "mysqlbinlog", "mysql.connect") } // Compile-time check. diff --git a/internal/driver/postgres/incremental.go b/internal/driver/postgres/incremental.go new file mode 100644 index 0000000..7532bae --- /dev/null +++ b/internal/driver/postgres/incremental.go @@ -0,0 +1,140 @@ +package postgres + +import ( + "context" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "sort" + "strconv" + "strings" + + "github.com/oklog/ulid/v2" + + "github.com/nixrajput/siphon/internal/driver" + "github.com/nixrajput/siphon/internal/errs" +) + +// IncrementalBaseInfo records what a base backup captured so a later +// incremental can resume from the correct WAL position. It is serialized into +// the dump Envelope (WALStart/WALEnd) and the slot is dropped when the chain +// is sealed (or via an orphan scan). +type IncrementalBaseInfo struct { + WALStart string // server LSN when the base backup began + WALEnd string // server LSN when the base backup finished + SlotName string // temporary physical replication slot anchoring WAL retention +} + +// CreateBaseSlot creates a temporary physical replication slot and records the +// start LSN. Call this immediately before taking a base backup; the slot +// prevents the server from recycling WAL the future incremental will need. +func (c *Conn) CreateBaseSlot(ctx context.Context) (*IncrementalBaseInfo, error) { + // Postgres replication slot names allow only [a-z0-9_]; ULIDs are Crockford + // base32 (uppercase), so lowercase it or pg_create_physical_replication_slot + // rejects the name with SQLSTATE 42602 ("contains invalid character"). + slot := "siphon_" + strings.ToLower(ulid.Make().String()) + // temporary=false so the slot survives this session (the incremental runs + // in a later session); we drop it explicitly via DropSlot. + if _, err := c.db.ExecContext(ctx, + "SELECT pg_create_physical_replication_slot($1, false, false)", slot); err != nil { + return nil, &errs.Error{Op: "postgres.create_slot", Code: errs.CodeSystem, Cause: err} + } + info := &IncrementalBaseInfo{SlotName: slot} + if err := c.db.QueryRowContext(ctx, "SELECT pg_current_wal_lsn()::text").Scan(&info.WALStart); err != nil { + _, _ = c.db.ExecContext(ctx, "SELECT pg_drop_replication_slot($1)", slot) // best-effort cleanup + return nil, &errs.Error{Op: "postgres.capture_lsn", Code: errs.CodeSystem, Cause: err} + } + return info, nil +} + +// CaptureBaseEnd records the end-of-base LSN into info. +func (c *Conn) CaptureBaseEnd(ctx context.Context, info *IncrementalBaseInfo) error { + if err := c.db.QueryRowContext(ctx, "SELECT pg_current_wal_lsn()::text").Scan(&info.WALEnd); err != nil { + return &errs.Error{Op: "postgres.capture_lsn", Code: errs.CodeSystem, Cause: err} + } + return nil +} + +// incrementalArgs builds the pg_receivewal argv to stream WAL from the slot +// into a destination DIRECTORY. pg_receivewal requires -D ; it does NOT +// support "-D -" (stdout). The slot (created at base time) anchors retention, +// so streaming from it captures WAL accumulated since the base. --endpos stops +// the stream at a defined LSN so the invocation terminates deterministically; +// --no-loop controls connection-RETRY behavior (don't loop reconnecting on a +// dropped connection), NOT WAL-end termination. This path needs validation +// against a live wal_level>=replica server (see incremental_test.go) — it is +// structurally complete but unproven locally (no Docker in this environment). +func incrementalArgs(p driver.Profile, slotName, dir, endpos string) []string { + return []string{ + "-h", p.Host, + "-p", strconv.Itoa(p.Port), + "-U", p.User, + "-D", dir, // pg_receivewal writes WAL segments into this directory + "--slot=" + slotName, + "--endpos=" + endpos, // stop at this LSN instead of streaming forever + "--synchronous", + "--no-loop", // do not retry the connection if it drops + "--verbose", + } +} + +// BackupIncremental streams WAL from the base's slot into a temp directory up +// to the server's current end LSN, then concatenates the resulting WAL segment +// files into w in name order. The caller prepends the dump Envelope at the app +// layer. +// +// pg_receivewal cannot stream to stdout (no "-D -"), so it must write segment +// files to a directory; we capture the current end LSN via pg_current_wal_lsn() +// and pass it as --endpos so the stream terminates at a defined point. This +// streaming path is exercised only in CI / against a live server — it is not +// validated locally (no Docker in this environment). +func (c *Conn) BackupIncremental(ctx context.Context, info IncrementalBaseInfo, w io.Writer) error { + var endpos string + if err := c.db.QueryRowContext(ctx, "SELECT pg_current_wal_lsn()::text").Scan(&endpos); err != nil { + return &errs.Error{Op: "postgres.backup_incremental", Code: errs.CodeSystem, Cause: err} + } + + tmpDir, err := os.MkdirTemp("", "siphon-wal-*") + if err != nil { + return &errs.Error{Op: "postgres.backup_incremental", Code: errs.CodeSystem, Cause: err} + } + defer func() { _ = os.RemoveAll(tmpDir) }() + + cmd := exec.CommandContext(ctx, "pg_receivewal", incrementalArgs(c.p, info.SlotName, tmpDir, endpos)...) + cmd.Env = append(os.Environ(), "PGPASSWORD="+c.p.Password) + cmd.Stderr = os.Stderr // surface pg_receivewal diagnostics (matches restore.go convention) + if err := cmd.Run(); err != nil { + return &errs.Error{Op: "postgres.backup_incremental", Code: errs.CodeSystem, Cause: err} + } + + // Copy the captured WAL segments to w in name order. Segment file names are + // fixed-width hex, so lexical sort is chronological order. + segments, err := filepath.Glob(filepath.Join(tmpDir, "*")) + if err != nil { + return &errs.Error{Op: "postgres.backup_incremental", Code: errs.CodeSystem, Cause: err} + } + sort.Strings(segments) + for _, seg := range segments { + f, err := os.Open(seg) + if err != nil { + return &errs.Error{Op: "postgres.backup_incremental", Code: errs.CodeSystem, Cause: err} + } + if _, err := io.Copy(w, f); err != nil { + _ = f.Close() + return &errs.Error{Op: "postgres.backup_incremental", Code: errs.CodeSystem, Cause: err} + } + _ = f.Close() + } + return nil +} + +// DropSlot removes the replication slot once a chain is sealed (or via an +// orphan scan on startup). Safe to call best-effort. +func (c *Conn) DropSlot(ctx context.Context, slotName string) error { + if _, err := c.db.ExecContext(ctx, "SELECT pg_drop_replication_slot($1)", slotName); err != nil { + return fmt.Errorf("drop slot %s: %w", slotName, err) + } + return nil +} diff --git a/internal/driver/postgres/incremental_test.go b/internal/driver/postgres/incremental_test.go new file mode 100644 index 0000000..36a794e --- /dev/null +++ b/internal/driver/postgres/incremental_test.go @@ -0,0 +1,51 @@ +//go:build integration + +package postgres + +import ( + "bytes" + "context" + "testing" + "time" + + "github.com/nixrajput/siphon/internal/driver" +) + +// TestIncremental_SlotAndLSNCapture verifies the slot lifecycle + LSN capture +// against a real Postgres container: create slot, take a base backup, record +// start/end LSN, then drop the slot. (WAL streaming via pg_receivewal needs a +// wal_level>=replica server and is exercised separately.) +func TestIncremental_SlotAndLSNCapture(t *testing.T) { + prof, cleanup, _ := startPostgres(t) + t.Cleanup(cleanup) + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + conn, err := Driver{}.Connect(ctx, prof) + if err != nil { + t.Fatalf("Connect: %v", err) + } + defer func() { _ = conn.Close() }() + pg := conn.(*Conn) + + info, err := pg.CreateBaseSlot(ctx) + if err != nil { + t.Fatalf("CreateBaseSlot: %v", err) + } + defer func() { _ = pg.DropSlot(ctx, info.SlotName) }() + + var base bytes.Buffer + if err := conn.Backup(ctx, driver.BackupOpts{}, &base); err != nil { + t.Fatalf("Backup base: %v", err) + } + if err := pg.CaptureBaseEnd(ctx, info); err != nil { + t.Fatalf("CaptureBaseEnd: %v", err) + } + if info.WALStart == "" || info.WALEnd == "" { + t.Fatalf("expected non-empty WAL LSNs, got %+v", info) + } + if info.SlotName == "" { + t.Fatalf("expected slot name") + } +} diff --git a/internal/dumps/chain.go b/internal/dumps/chain.go new file mode 100644 index 0000000..62c823a --- /dev/null +++ b/internal/dumps/chain.go @@ -0,0 +1,42 @@ +package dumps + +import ( + "errors" + "fmt" +) + +// ResolveChain returns the ordered list of Metas that, applied in order, +// reconstruct targetID. Element 0 is the base; the last element is the target. +// +// 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) { + visited := map[string]bool{} + var chain []Meta + + cur := targetID + for { + if visited[cur] { + return nil, fmt.Errorf("chain cycle detected at %s", cur) + } + visited[cur] = true + + m, err := c.ReadMeta(cur) + if err != nil { + return nil, fmt.Errorf("chain broken at %s: %w", cur, err) + } + chain = append([]Meta{*m}, chain...) // prepend + + // Reached the base: self-referential BaseID, or legacy empty BaseID. + if m.BaseID == "" || m.BaseID == m.ID { + return chain, nil + } + + // Mid-chain incremental must point at a parent. + if m.ParentID == "" { + return nil, errors.New("chain broken: incremental " + m.ID + " has no parent_id") + } + cur = m.ParentID + } +} diff --git a/internal/dumps/chain_test.go b/internal/dumps/chain_test.go new file mode 100644 index 0000000..8ba60ad --- /dev/null +++ b/internal/dumps/chain_test.go @@ -0,0 +1,69 @@ +package dumps + +import ( + "testing" + "time" +) + +func TestResolveChain_SingleBase(t *testing.T) { + c, _ := NewCatalog(t.TempDir()) + base := &Meta{ID: "base", BaseID: "base", Created: time.Now()} + _ = c.WriteMeta(base) + + chain, err := c.ResolveChain("base") + if err != nil { + t.Fatal(err) + } + if len(chain) != 1 || chain[0].ID != "base" { + t.Fatalf("got %v; want [base]", chain) + } +} + +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()}) + + chain, err := c.ResolveChain("inc1") + if err != nil { + t.Fatal(err) + } + if len(chain) != 2 || chain[0].ID != "base" || chain[1].ID != "inc1" { + t.Fatalf("got %v; want [base, inc1]", chain) + } +} + +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()}) + + chain, err := c.ResolveChain("inc2") + if err != nil { + t.Fatal(err) + } + if len(chain) != 3 || chain[0].ID != "base" || chain[1].ID != "inc1" || chain[2].ID != "inc2" { + t.Fatalf("got %v; want [base, inc1, inc2]", chain) + } +} + +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"}) + + if _, err := c.ResolveChain("a"); err == nil { + t.Fatal("expected cycle error") + } +} + +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()}) + + if _, err := c.ResolveChain("inc1"); err == nil { + t.Fatal("expected broken-chain error (missing base)") + } +} diff --git a/internal/dumps/envelope.go b/internal/dumps/envelope.go new file mode 100644 index 0000000..4e1cb24 --- /dev/null +++ b/internal/dumps/envelope.go @@ -0,0 +1,89 @@ +package dumps + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "time" +) + +const ( + EnvelopeMagic = "SIPH" + EnvelopeSize = 4096 +) + +type EnvelopeType string + +const ( + EnvelopeBase EnvelopeType = "base" + EnvelopeIncremental EnvelopeType = "incremental" +) + +// Envelope is the 4 KB JSON header prepended to every dump. The native +// dump bytes follow immediately after. +type Envelope struct { + Siphon string `json:"siphon"` + Type EnvelopeType `json:"type"` + Driver string `json:"driver"` + EngineVersion string `json:"engine_version,omitempty"` + BaseID string `json:"base_id,omitempty"` + ParentID string `json:"parent_id,omitempty"` + WALStart string `json:"wal_start,omitempty"` + WALEnd string `json:"wal_end,omitempty"` + BinlogFile string `json:"binlog_file,omitempty"` + BinlogStart uint64 `json:"binlog_start,omitempty"` + BinlogEnd uint64 `json:"binlog_end,omitempty"` + Checksum string `json:"checksum,omitempty"` + Tables []TableEntry `json:"tables,omitempty"` + Created time.Time `json:"created"` +} + +var ErrInvalidEnvelope = errors.New("invalid siphon envelope") + +// WriteEnvelope writes a padded 4 KB header to w. Returns the number of +// bytes written (always 4096 on success). +func WriteEnvelope(w io.Writer, e *Envelope) (int, error) { + if e.Siphon == "" { + e.Siphon = "1.0" + } + if e.Created.IsZero() { + e.Created = time.Now().UTC() + } + body, err := json.Marshal(e) + if err != nil { + return 0, err + } + if 4+len(body) > EnvelopeSize-1 { + return 0, fmt.Errorf("envelope JSON is %d bytes; max %d", len(body), EnvelopeSize-5) + } + + buf := make([]byte, EnvelopeSize) + copy(buf[0:4], []byte(EnvelopeMagic)) + copy(buf[4:], body) + for i := 4 + len(body); i < EnvelopeSize-1; i++ { + buf[i] = ' ' + } + buf[EnvelopeSize-1] = '\n' + return w.Write(buf) +} + +// ReadEnvelope reads and validates the 4 KB header from r. Returns the +// parsed Envelope and a reader positioned at the start of the native +// dump bytes. +func ReadEnvelope(r io.Reader) (*Envelope, io.Reader, error) { + header := make([]byte, EnvelopeSize) + if _, err := io.ReadFull(r, header); err != nil { + return nil, nil, fmt.Errorf("%w: short read", ErrInvalidEnvelope) + } + if string(header[0:4]) != EnvelopeMagic { + return nil, nil, fmt.Errorf("%w: missing magic", ErrInvalidEnvelope) + } + body := bytes.TrimRight(header[4:EnvelopeSize-1], " ") + e := &Envelope{} + if err := json.Unmarshal(body, e); err != nil { + return nil, nil, fmt.Errorf("%w: %v", ErrInvalidEnvelope, err) + } + return e, r, nil +} diff --git a/internal/dumps/envelope_test.go b/internal/dumps/envelope_test.go new file mode 100644 index 0000000..d2571c0 --- /dev/null +++ b/internal/dumps/envelope_test.go @@ -0,0 +1,58 @@ +package dumps + +import ( + "bytes" + "strings" + "testing" + "time" +) + +func TestEnvelope_Roundtrip(t *testing.T) { + in := &Envelope{ + Siphon: "1.0", + Type: EnvelopeBase, + Driver: "postgres", + Created: time.Now().UTC().Truncate(time.Second), + } + var buf bytes.Buffer + n, err := WriteEnvelope(&buf, in) + if err != nil { + t.Fatalf("WriteEnvelope: %v", err) + } + if n != EnvelopeSize { + t.Fatalf("wrote %d bytes; want %d", n, EnvelopeSize) + } + buf.WriteString("native-dump-bytes") + + out, body, err := ReadEnvelope(&buf) + if err != nil { + t.Fatalf("ReadEnvelope: %v", err) + } + if out.Driver != "postgres" || out.Type != EnvelopeBase { + t.Fatalf("Envelope round-trip mismatch: %+v", out) + } + + rest := make([]byte, 64) + n2, _ := body.Read(rest) + if !strings.HasPrefix(string(rest[:n2]), "native-dump-bytes") { + t.Fatalf("body reader misaligned; got %q", rest[:n2]) + } +} + +func TestEnvelope_MissingMagic(t *testing.T) { + junk := make([]byte, EnvelopeSize) + copy(junk, []byte("NOPE")) + _, _, err := ReadEnvelope(bytes.NewReader(junk)) + if err == nil { + t.Fatal("expected error on bad magic") + } +} + +func TestEnvelope_OversizedJSON(t *testing.T) { + huge := strings.Repeat("x", EnvelopeSize) + e := &Envelope{Siphon: "1.0", Driver: "postgres", EngineVersion: huge} + _, err := WriteEnvelope(&bytes.Buffer{}, e) + if err == nil { + t.Fatal("expected error on oversized envelope") + } +} diff --git a/internal/jobs/stream.go b/internal/jobs/stream.go new file mode 100644 index 0000000..b6e2033 --- /dev/null +++ b/internal/jobs/stream.go @@ -0,0 +1,193 @@ +package jobs + +import ( + "context" + "io" + "sync" + "sync/atomic" +) + +// Stream is a bounded in-memory pipe satisfying io.Reader, io.Writer, and +// io.Closer. It replaces io.Pipe for streaming sync so backpressure is +// observable: FillPercent reports how full the buffer is (a metric a job-panel +// view can surface; not yet rendered in the TUI). Safe for one writer goroutine +// + one reader goroutine, and Close may be called from any goroutine (e.g. +// CloseOnCtx on cancel). +// +// Close never closes the buffer channel s.ch — closing a channel concurrently +// with a send is itself a data race (and a send on a closed channel panics). +// Instead Close closes only the s.done signal; Write selects on done so it can +// never send after close, and Read selects on both ch and done so it drains +// remaining chunks and then reports EOF. This keeps Stream race-free even when +// Write (writer goroutine) and Close (CloseOnCtx goroutine) run concurrently. +type Stream struct { + ch chan []byte + capChunks int + used atomic.Int64 + done chan struct{} // closed by Close/CloseErr; signals writer-done + closeErr atomic.Pointer[error] // set before done is closed by CloseErr; nil means clean EOF + closeOnce sync.Once + overflow []byte // leftover from the previous Read +} + +// NewStream creates a Stream buffering up to capChunks chunks (each an +// independent []byte). capChunks <= 0 defaults to 64 (~64MB at 1MB/chunk): +// enough for a laptop, small enough that a slow target shows visible pressure. +func NewStream(capChunks int) *Stream { + if capChunks <= 0 { + capChunks = 64 + } + return &Stream{ + ch: make(chan []byte, capChunks), + capChunks: capChunks, + done: make(chan struct{}), + } +} + +// maxChunk bounds the size of a single buffered chunk. A large Write is split +// into pieces of at most maxChunk bytes so the bounded channel actually bounds +// memory to ~capChunks×maxChunk; otherwise one huge Write would enqueue the +// whole slice as a single unbounded chunk. +const maxChunk = 1 << 20 // 1 MiB + +// Write enqueues a copy of p, splitting it into chunks of at most maxChunk +// bytes so the bounded channel bounds memory. Blocks when the buffer is full. +// Returns io.ErrClosedPipe if the stream is closed. Because s.ch is never +// closed, the send can never panic; the select simply abandons the send and +// returns io.ErrClosedPipe if Close fires while Write is parked. Per the +// io.Writer contract, a short write returns n < len(p) together with the error. +func (s *Stream) Write(p []byte) (int, error) { + // Fast path: reject if already closed before allocating a copy. + select { + case <-s.done: + return 0, io.ErrClosedPipe + default: + } + written := 0 + for written < len(p) { + end := written + maxChunk + if end > len(p) { + end = len(p) + } + piece := p[written:end] + cp := make([]byte, len(piece)) + copy(cp, piece) + select { + case <-s.done: + return written, io.ErrClosedPipe + case s.ch <- cp: + s.used.Add(1) + written += len(piece) + } + } + return written, nil +} + +// Read drains the buffer. Buffered chunks are always served first (even after +// Close), so no data is lost; once the writer has Closed and the buffer is +// empty, Read returns io.EOF. +func (s *Stream) Read(p []byte) (int, error) { + // io.Reader contract: a zero-length read returns (0, nil) immediately and + // must not block waiting for data. + if len(p) == 0 { + return 0, nil + } + if len(s.overflow) > 0 { + n := copy(p, s.overflow) + s.overflow = s.overflow[n:] + return n, nil + } + // Prefer a buffered chunk if one is ready, regardless of done state, so + // Close does not drop already-buffered data. + select { + case chunk := <-s.ch: + return s.deliver(p, chunk), nil + default: + } + // Buffer momentarily empty: block until a chunk arrives or the writer is + // done. If done fires, drain any final chunk that raced in, else EOF. + select { + case chunk := <-s.ch: + return s.deliver(p, chunk), nil + case <-s.done: + select { + case chunk := <-s.ch: + return s.deliver(p, chunk), nil + default: + return 0, s.endErr() + } + } +} + +// endErr returns the error supplied to CloseErr, or io.EOF for a clean Close. +// Read by Read only after observing s.done closed, so the closeErr store (which +// happens before close(s.done) in CloseErr) is guaranteed visible. +func (s *Stream) endErr() error { + if p := s.closeErr.Load(); p != nil { + return *p + } + return io.EOF +} + +// deliver copies chunk into p, stashes any remainder as overflow, and +// decrements the buffered-chunk counter. +func (s *Stream) deliver(p, chunk []byte) int { + s.used.Add(-1) + n := copy(p, chunk) + if n < len(chunk) { + s.overflow = chunk[n:] + } + return n +} + +// Close marks the writer side done; subsequent Writes return io.ErrClosedPipe +// and readers see EOF after draining buffered chunks. It does NOT close s.ch +// (that would race a concurrent Write), so Close is safe from any goroutine. +// Idempotent. +func (s *Stream) Close() error { + s.closeOnce.Do(func() { + close(s.done) + }) + return nil +} + +// CloseErr is like Close but causes Read to return err (instead of io.EOF) +// once buffered chunks are drained — the bounded-buffer analogue of +// io.PipeWriter.CloseWithError. A producer uses it to propagate a failure +// (e.g. a truncated backup) to the consumer as a read error, so the consumer +// doesn't mistake a partial stream for a clean end. A nil err behaves like +// Close. Idempotent; the first Close/CloseErr wins. The error is stored before +// s.done is closed, so any Read that observes done also observes the error. +func (s *Stream) CloseErr(err error) error { + s.closeOnce.Do(func() { + if err != nil { + s.closeErr.Store(&err) + } + close(s.done) + }) + return nil +} + +// FillPercent returns 0..100 = buffered chunks / capacity. Observable backpressure. +func (s *Stream) FillPercent() int { + if s.capChunks == 0 { + return 0 + } + pct := int(s.used.Load() * 100 / int64(s.capChunks)) + if pct < 0 { + pct = 0 + } + if pct > 100 { + pct = 100 + } + return pct +} + +// CloseOnCtx closes the stream when ctx is done — convenience for producer +// goroutines so a cancelled job tears the pipe down. +func (s *Stream) CloseOnCtx(ctx context.Context) { + go func() { + <-ctx.Done() + _ = s.Close() + }() +} diff --git a/internal/jobs/stream_test.go b/internal/jobs/stream_test.go new file mode 100644 index 0000000..626a0b1 --- /dev/null +++ b/internal/jobs/stream_test.go @@ -0,0 +1,269 @@ +package jobs + +import ( + "context" + "errors" + "io" + "sync" + "testing" + "time" +) + +// TestStream_CloseErr_PropagatesError locks in the failure-propagation contract +// the streaming sync relies on: after CloseErr(err), Read must serve every +// buffered chunk and then return that exact err (not io.EOF). A consumer +// (Restore) thus sees a read error on a truncated stream instead of mistaking +// it for a clean end — the bounded-buffer analogue of pipe CloseWithError. +func TestStream_CloseErr_PropagatesError(t *testing.T) { + wantErr := errors.New("backup failed mid-dump") + s := NewStream(8) + const n = 3 + for i := 0; i < n; i++ { + if _, err := s.Write([]byte("data")); err != nil { + t.Fatalf("Write %d: %v", i, err) + } + } + _ = s.CloseErr(wantErr) + + // Drain: buffered chunks first (each a clean read), then the final error. + buf := make([]byte, 64) + var got int + var finalErr error + for { + nRead, err := s.Read(buf) + got += nRead + if err != nil { + finalErr = err + break + } + } + if want := n * len("data"); got != want { + t.Fatalf("drained %d bytes before error; want %d (chunks dropped)", got, want) + } + if !errors.Is(finalErr, wantErr) { + t.Fatalf("final Read error = %v; want %v", finalErr, wantErr) + } + if errors.Is(finalErr, io.EOF) { + t.Fatalf("final Read error must not be io.EOF on a failed stream") + } +} + +// TestStream_CloseNilErr_YieldsEOF confirms a clean Close (and CloseErr(nil)) +// still ends with io.EOF — a nil error is not a failure. +func TestStream_CloseNilErr_YieldsEOF(t *testing.T) { + s := NewStream(4) + if _, err := s.Write([]byte("ok")); err != nil { + t.Fatalf("Write: %v", err) + } + _ = s.CloseErr(nil) // nil err behaves like Close + + out, err := io.ReadAll(s) // ReadAll treats io.EOF as success + if err != nil { + t.Fatalf("ReadAll after CloseErr(nil): %v; want clean EOF", err) + } + if string(out) != "ok" { + t.Fatalf("got %q; want %q", out, "ok") + } +} + +func TestStream_WriteRead_Roundtrip(t *testing.T) { + s := NewStream(4) + payload := []byte("hello-buffered-world") + go func() { + _, _ = s.Write(payload) + _ = s.Close() + }() + out, err := io.ReadAll(s) + if err != nil { + t.Fatal(err) + } + if string(out) != string(payload) { + t.Fatalf("got %q; want %q", out, payload) + } +} + +// TestStream_DrainsAllChunksBeforeEOF locks in the critical invariant: when +// Close fires while chunks are still buffered, every buffered chunk must be +// delivered before io.EOF. A naive select{ch; done} in Read would randomly +// return EOF and drop the tail — silently truncating a streamed dump. +func TestStream_DrainsAllChunksBeforeEOF(t *testing.T) { + const n = 50 + s := NewStream(64) // cap > n so all writes buffer without a reader + for i := 0; i < n; i++ { + if _, err := s.Write([]byte("chunk")); err != nil { + t.Fatalf("Write %d: %v", i, err) + } + } + _ = s.Close() // close with all n chunks still buffered + + out, err := io.ReadAll(s) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if got, want := len(out), n*len("chunk"); got != want { + t.Fatalf("drained %d bytes after Close; want %d (chunks dropped before EOF)", got, want) + } +} + +func TestStream_FillPercent_TracksWrites(t *testing.T) { + s := NewStream(4) + // Fill without a reader so chunks accumulate (cap 4, write 3 -> ~75%). + for i := 0; i < 3; i++ { + if _, err := s.Write([]byte("chunk")); err != nil { + t.Fatalf("Write: %v", err) + } + } + got := s.FillPercent() + if got <= 0 || got > 100 { + t.Fatalf("FillPercent = %d; want 1..100 after 3/4 writes", got) + } +} + +func TestStream_FillPercent_ClampedAndEmpty(t *testing.T) { + s := NewStream(2) + if got := s.FillPercent(); got != 0 { + t.Fatalf("FillPercent on empty = %d; want 0", got) + } + for i := 0; i < 2; i++ { + if _, err := s.Write([]byte("x")); err != nil { + t.Fatalf("Write: %v", err) + } + } + if got := s.FillPercent(); got != 100 { + t.Fatalf("FillPercent full = %d; want 100", got) + } +} + +func TestStream_WriteAfterClose_ReturnsErrClosedPipe(t *testing.T) { + s := NewStream(2) + _ = s.Close() + if _, err := s.Write([]byte("x")); err != io.ErrClosedPipe { + t.Fatalf("err = %v; want io.ErrClosedPipe", err) + } +} + +func TestStream_CloseIsIdempotent(t *testing.T) { + s := NewStream(2) + if err := s.Close(); err != nil { + t.Fatalf("first Close: %v", err) + } + if err := s.Close(); err != nil { // must not panic on double close + t.Fatalf("second Close: %v", err) + } +} + +func TestStream_PartialRead_Overflow(t *testing.T) { + s := NewStream(2) + if _, err := s.Write([]byte("abcdef")); err != nil { + t.Fatalf("Write: %v", err) + } + _ = s.Close() + buf := make([]byte, 4) + n, err := s.Read(buf) + if err != nil { + t.Fatalf("Read: %v", err) + } + if string(buf[:n]) != "abcd" { + t.Fatalf("first read = %q; want %q", buf[:n], "abcd") + } + rest, err := io.ReadAll(s) + if err != nil { + t.Fatalf("ReadAll rest: %v", err) + } + if string(rest) != "ef" { + t.Fatalf("rest = %q; want %q", rest, "ef") + } +} + +func TestStream_ConcurrentWriteCloseNoPanic(t *testing.T) { + // Stress the Write/Close race directly (run under -race). A concurrent + // Close must never panic the writer; Write must return without sending + // on a closed channel. Join every goroutine so the stressed interleavings + // actually complete before the test returns (otherwise the test can pass + // before any race window is exercised). + var wg sync.WaitGroup + for i := 0; i < 200; i++ { + s := NewStream(1) + wg.Add(3) + go func() { defer wg.Done(); _ = s.Close() }() + go func() { defer wg.Done(); _, _ = s.Write([]byte("x")) }() + go func() { + defer wg.Done() + buf := make([]byte, 8) + _, _ = s.Read(buf) + }() + } + wg.Wait() +} + +// TestStream_ZeroLenRead_DoesNotBlock confirms the io.Reader contract: a Read +// with a zero-length buffer returns (0, nil) immediately, even on an open and +// empty stream where a blocking select would otherwise hang. +func TestStream_ZeroLenRead_DoesNotBlock(t *testing.T) { + s := NewStream(4) // open, empty + done := make(chan struct{}) + go func() { + n, err := s.Read(make([]byte, 0)) + if n != 0 || err != nil { + t.Errorf("Read(empty) = (%d, %v); want (0, nil)", n, err) + } + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("zero-length Read blocked on an open empty stream") + } +} + +// TestStream_LargeWrite_SplitsIntoChunks confirms a single Write larger than +// maxChunk is split into multiple buffered chunks (so the bounded channel +// actually bounds memory), reflected by FillPercent rising above one chunk's +// worth. +func TestStream_LargeWrite_SplitsIntoChunks(t *testing.T) { + const cap = 64 + s := NewStream(cap) + // 4 MiB write => 4 chunks at 1 MiB each, all buffered without a reader. + big := make([]byte, 4*maxChunk) + n, err := s.Write(big) + if err != nil { + t.Fatalf("Write: %v", err) + } + if n != len(big) { + t.Fatalf("Write returned %d; want %d", n, len(big)) + } + // 4 buffered chunks out of cap 64 => 4*100/64 = 6%. + if got, want := s.FillPercent(), 4*100/cap; got != want { + t.Fatalf("FillPercent = %d; want %d (one big Write must buffer 4 chunks)", got, want) + } + // And it must round-trip intact once closed. + _ = s.Close() + out, err := io.ReadAll(s) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if len(out) != len(big) { + t.Fatalf("round-tripped %d bytes; want %d", len(out), len(big)) + } +} + +func TestStream_CloseOnCtx(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + s := NewStream(64) + s.CloseOnCtx(ctx) + cancel() + + // Poll (bounded) until CloseOnCtx has closed the stream after cancel. + // Use a fresh single Write each iteration; cap 64 means a Write before + // close lands buffers without blocking. + deadline := time.Now().Add(time.Second) + for { + if _, err := s.Write([]byte("x")); err == io.ErrClosedPipe { + return // CloseOnCtx closed the stream after cancel. + } + if time.Now().After(deadline) { + t.Fatal("stream not closed within 1s of ctx cancel") + } + time.Sleep(time.Millisecond) + } +}