From 361f5140a48d5611f12ff690ea578018cfa58250 Mon Sep 17 00:00:00 2001 From: pingqiu Date: Wed, 29 Apr 2026 23:38:29 -0700 Subject: [PATCH] =?UTF-8?q?sw-block/design:=20v3-recovery-wal-shipper-mini?= =?UTF-8?q?-plan=20+=20=C2=A710=20P2d=20decision=20request?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the WalShipper implementation mini-plan that bridges v3-recovery-wal-shipper-spec.md to the seaweed_block layout (phased PR rollout P0..P4, INV ↔ test mapping, reviewer checklist). §10 P2d decision request — the architect-gated handoff: P2c is closed (slice A / B-1 / B-2 merged on g7-redo/wal-shipper-impl). The bridging senderBacklogSink owns the live-write buffer + flushAndSeal under sinkMu; Sender.Run barriers as soon as sink.DrainBacklog returns; Close/closeCh/liveQueue/drainAndSeal are deleted from Sender. Atomic-seal contract migrates intact (capture-vs-reject from queueMu → sinkMu). P2d is gated on a three-axis decision the architect must make before a real transport.WalShipper sink can replace the bridging path: 1. Body format on the dual-lane port: (A) MsgShipEntry payload (unify on legacy steady encoding), OR (B) frameWALEntry payloads (teach WalShipper.Emit to encode), OR (C) documented third (e.g. envelope byte). 2. Single applier owner: recovery.Receiver vs transport replica handler. 3. Replay source of truth: which encoding the on-disk WAL playback decoder reads. §10 also lists pre-decision deliverables that can land in parallel: adapter scaffolding (transport-side struct satisfying recovery.WalShipperSink by duck typing) + integration tests for architect rules 1+2 (emit context before StartSession; restore steady lineage after EndSession). V2 wire-compat is gated separately per feedback_porting_discipline.md. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../v3-recovery-wal-shipper-mini-plan.md | 182 ++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 sw-block/design/v3-recovery-wal-shipper-mini-plan.md diff --git a/sw-block/design/v3-recovery-wal-shipper-mini-plan.md b/sw-block/design/v3-recovery-wal-shipper-mini-plan.md new file mode 100644 index 000000000..c1f671c4d --- /dev/null +++ b/sw-block/design/v3-recovery-wal-shipper-mini-plan.md @@ -0,0 +1,182 @@ +# WalShipper implementation mini-plan (**PR bridge**) + +**Status**: Draft — executable sequence from **`v3-recovery-wal-shipper-spec.md`** to **`seaweed_block`** concrete diffs. +**Supersedes narrative scope of**: **`v3-recovery-unified-wal-stream-mini-plan.md` §3.2 (#3 unified stream implementation)** — that track treated recovery `sender` as the Wal scheduler; **non-compliant** with consensus **§6** + wal-shipper-spec **INV-NO-DOUBLE-LIVE**. Architect may add formal **SUPERSEDED** banners on kickoff/unified docs; **this file does not block on that.** + +**Normative**: **`v3-recovery-wal-shipper-spec.md`** (**§2–§§7**) + **`v3-recovery-algorithm-consensus.md`** **§I / §II §6**. + +**Anti‑archive branch**: **`g7-redo/unified-wal-impl`** — **do not PR** as WalShipper truth; cherry-pick only **tests/fixtures** if still valid after refactor audit. + +--- + +## 0. Outcome (**definition of done**) + +1. **One Wal emit decision stream per `(volume identity, replicaID)`** on Primary — **cursor + shipMu + spec §4**. +2. **Steady path** (`ReplicaPeer` / **`BlockExecutor.Ship`**) and **recover Wal path** (**dual‑lane frames**) **both** funnel through **that** scheduler — **`INV-NO-DOUBLE-LIVE`**. +3. **`core/recovery/sender.go`** (**`Sender`**) is **thin**: session bracket, **base lane** (`streamBase`/barrier taxonomy), **`Run` choreography** — **no** independent `cursor` / `streamUntilHead` competing with **`Ship()`**. +4. **`R1` double‑check** and **`R2` lag hook** present per **spec §§5–6**. +5. **`§9` tests** mapped below — **`CHK-WALSHIPPER-SINGLE-CURSOR` / `NO-GAP-R1` proven at unit/concurrency scope** — *not* as the sole reliance on flaky multi‑second integration (architect constraint). + +--- + +## 1. Where code lives (**seaweed_block** anchors) + +| Component | Primary file(s) (**today**) | Target responsibility | +|-----------|----------------------------|------------------------| +| **Stateful WalShipper** | *new* **`core/transport/wal_shipper.go`** | Holds **`cursor`**, **`shipMu`**, session **`fromLSN`**, **`NotifyAppend`** / **`DrainBacklog`**, **`OnShipTimer`**. Emits via injected **`EmitFunc`** (mock in P0) | +| **Steady live ship** | **`core/transport/ship_sender.go`** — **`BlockExecutor.Ship`**| **Delegates emit** to per‑replica **WalShipper** (same mutex family as append path — **INV-SINGLE**) — **never** standalone encode+write concurrent with backlog pump | +| **Executor registry** | **`core/transport/executor.go`** **`sessions`** map | Registers **`WalShipper` per replica** keyed **stable ID** (**volume + ReplicaID**) — satisfies **INV-SINGLE** dedup (**spec §2 forbids** dual goroutines advancing cursor) | +| **Recovery orchestration** | **`core/recovery/sender.go`**| Drops independent Wal pump → **`WalShipper.DrainBacklog(fromLSN, ctx)`** (or **`BeginRecoverSession`** + **cursor reset**) — skeleton P0; wired P2 | +| **Local Wal append ingress (**steady**)** | **Chain below** terminates in **`core/transport/ship_sender.go`::`BlockExecutor.Ship`** (**P1**)| **Recommended hook**: **`Ship` delegates** to **`WalShipper.NotifyAppend`** (or **`EmitAccordingTo§4`**). **`ReplicationVolume.OnLocalWrite` / `ReplicaPeer.ShipEntry` signatures unchanged** | + +### 1.1 Steady-path write fan‑out (**`seaweed_block`**, verified) + +**Call chain (4 hops, **P0** leaves this chain untouched):** + +``` +core/replication/volume.go:421 ReplicationVolume.OnLocalWrite(...) + → iterates v.peers, peer.ShipEntry(...) per replica + +core/replication/peer.go:459 ReplicaPeer.ShipEntry(...) + → state gate, builds peer lineage + +core/replication/peer.go:480 p.executor.Ship(p.target.ReplicaID, peerLineage, lba, lsn, data) + +core/transport/ship_sender.go:58 BlockExecutor.Ship(...) + → session lookup, lazy-dial, WriteMsg(MsgShipEntry / SWRP) +``` + +**P1 delegation rule**: add **`walShippers map[string]*WalShipper`** keyed by **`ReplicaID`** (or richer key if volumes share executor — reviewer confirms). **`BlockExecutor.Ship(...)` → `WalShipper` emits** via injected **`EmitFunc`** encoding **`MsgShipEntry`** — single live wire path (**INV‑NO‑DOUBLE‑LIVE** prelude). + +**P0 note**: **`WalShipper`** type lives in **`wal_shipper.go`** with mocks only — **does not alter** this chain until **P1**. + +*§2 design choice (**`shipMu`** vs actor) applies where **`WalShipper`** merges **`NotifyAppend`** / timer / drain.* + +--- + +## 2. Architectural choice (**serialize append vs ship**) + +Pick **before coding** (mini‑plan reviewer sign‑off): + +- **Choice A (`shipMu`)** — **Broad `shipMu`**: append callback that notifies shipper acquires **`shipMu`** after `lsn` fixed, bumps **`head`**, runs **§4 emit loop** excerpt. *Simplest for R1 double‑check.* + +- **Choice B (actor queue)** — Single goroutine consumes `{Append, ShipOpportunity}` messages; **no `shipMu`** but **FIFO** proves **INV-SINGLE**. *Harder glue to existing `Ship` callers.* + +Default recommendation: **A** for **minimal moving parts** in first merge. + +--- + +## 3. Phased rollout (**merged PR chunks**) + +| Phase | Content | Receipt | +|-------|---------|---------| +| **P0** | Introduce **`WalShipper`** type + **`NewWalShipperForReplica(...)`**; **pure unit tests**: **INV-MONOTONIC-CURSOR**, **INV-SUBSET** (fake `head`/`emit` mocks) | Green package tests without executor | +| **P1** | **`BlockExecutor`** registry: create/destroy **`WalShipper`** with **`StartRebuild`/`EndSession`/steady attach**; **`Ship`** delegates **`Emit`** | Existing **`ship_sender_test` / replicated path** unchanged or extended | +| **P2** | **Drain API** wired from **`recovery.Sender`**; **delete**/disable **`recovery/sender`** duplicate pump | **`e2e_test` / stub** revived only if semantics unchanged | +| **P3** | **§5 R1** **`AssertCaughtUp...`** + **§6 `lag`/SignalSaturation** ( thresholds **from config/constants** ) | **Test‑R1-Flip-NoGap**: **stress** goroutine concurrent append (**unit**, `-race`) | +| **P4** | Cleanup dead code (**unified-wal-impl** ports), **`docs` supersede headers** (#architect tooling) | Consensus **G0** checklist ticked | + +Phases **P0–P1** safe to merge before full recovery deletes if **INV-NO-DOUBLE-LIVE** holds by construction (**only WalShipper calls wire write**). + +--- + +## 4. Spec ↔ invariant ↔ test (**§9 wal-shipper-spec expanded**) + +Wal-shipper-spec **§9** names coarse tests — here map **package + style**: + +| Spec ref | Minimal test (**unit first**) | +|----------|-------------------------------| +| **INV-SINGLE** | **`-race`**: concurrent **`NotifyAppend`** + **`Drain`** from N goroutines — **≤1 emitting** (`atomic` counter on `EmitFunc`) | +| **INV-MONOTONIC-CURSOR** | Table: never regress **`cursor`** without `ResetSession` | +| **INV-SUBSET** | Emit mock asserts **`pin < lsn ≤ head`** for every **`Emit`** | +| **INV-NO-GAP-R1** (**§5**) | Scripted: `cursor == head` at **`H0`**, interleave **`head++`**, assert **`Emit(H0+1)`** occurs | +| **INV-NO-DOUBLE-LIVE** (**CHK surrogate**) | **Two entrypoints**: call **`WalShipper`** + legacy **`Ship` mock** — after P1, **`Ship` must not bypass** (**single emit counter**) | +| **§6 `lag`/R2** | Fake clock or small threshold: **`SignalSaturation` called exactly once**, session hook receives reason | + +**Integration**: **smoke** only (daemon up, no assertion of **`CHK-*` solely from E2E**). + +--- + +## 5. **PR reviewer checklist** + +- [ ] **No duplicate `Emit` routes** — grep **`frameWALEntry`/`MsgShipEntry` write paths** reachable from **`recovery/Sender`** *and* **`Ship`** without **`WalShipper`**. +- [ ] **`R1`** procedure identifiable in code (**double read `head`** or equivalent proof in PR description). +- [ ] **`g7-redo/unified-wal-impl`** **NOT** landed as authoritative Wal scheduling. +- [ ] Consensus **`G1`**: **INV-NO-DOUBLE-LIVE** exercised **deterministically** (unit/table). + +--- + +## 6. Dependencies & parallelism + +| Item | Blocking? | +|------|-----------| +| Architect **SUPERSEDED** on unified kickoff | **No** — optional doc hygiene same or follow PR | +| **`v3-storage-logical-pin-gate.md`** (smartwal **`RecycleFloorGate`**) | **Orthogonal** — WalShipper refactor should not tighten substrate in same mega-PR | + +--- + +## 7. Estimate (**order of magnitude**, revisable) + +| Slice | Rough LOC | +|-------|-----------| +| **`wal_shipper.go` (+ tests P0)** | +250 prod + 300 test | +| **`ship_sender.go`/executor refactor** | +150 / −200 | +| **`recovery/sender.go` thin** | −200 / +80 | +| **R1/R2 + race tests** | +200 | + +--- + +## 8. Revision + +| Date | Change | +|------|--------| +| 2026-04-29 | v0.1 — Bridges **wal-shipper-spec** ↔ **`seaweed_block`**; supersede unified §3.2 #3; phased PR | +| 2026-04-29 | v0.2 — **§1.1** steady-path **four-hop fan-out** + **`Ship` delegation** (**P1**); **`DrainBacklog`** naming | +| 2026-04-29 | v0.3 — **§10 P2d decision request** appended; P2c split into **slice A / B-1 / B-2** (all merged into `g7-redo/wal-shipper-impl`) | + +--- + +## 9. Answer card (**for implementer FAQ**) + +| Question | Answer | +|----------|--------| +| **(a) mini-plan before code?** | **Yes (`this file`) —** spec lacks file/layout; avoids second wrong placement | +| **(b) wait for superseded banner?** | **No blocker** | +| **(c) skip mini-plan?** | **Not advised** — already slipped once (**recovery.sender==shipper**) | + +--- + +## 10. **P2d decision request** (architect-gated) + +**Status**: **P2c is closed** (slice A / B-1 / B-2 merged into `g7-redo/wal-shipper-impl`). Bridging **`senderBacklogSink`** owns the live-write buffer + `flushAndSeal` under `sinkMu`; `Sender.Run` barriers as soon as `sink.DrainBacklog` returns; `Close`/`closeCh`/`liveQueue`/`drainAndSeal` are deleted from `Sender`. Atomic-seal contract preserved (capture-vs-reject migrates from `queueMu` → `sinkMu`). + +**P2d cannot start** until the architect picks one canonical dual-lane wire body format and one applier owner. The bridging sink stays in place until then; once a real `transport.WalShipper` adapter replaces it, the buffer + flushAndSeal scaffolding here goes too. + +### 10.1 The decision (three-axis, narrowly scoped) + +| Axis | Options | Blast radius | +|------|---------|--------------| +| **Body format on the dual-lane port** | (A) **`MsgShipEntry`** payload (`EncodeShipEntry` from transport) — unify on legacy steady-ship encoding | recovery/`frameWALEntry` decode path on receiver must **delegate to** transport apply path *or* re-encode kind tagging into metadata; WAL-replay tooling that assumes backlog payload shape must align | +| | (B) **`frameWALEntry`** payloads (`encodeWALEntry` from recovery) — teach `WalShipper.EmitFunc` to encode this | `BlockExecutor` / `WalShipper` `EmitFunc` must encode `frameWALEntry`, not `MsgShipEntry`; existing `MsgShipEntry`-only parsers need explicit **lane vs message-type dispatch** | +| | (C) **Documented third** (e.g. envelope byte that disambiguates) | Both decoders gain a tag check; cleanest if neither current shape is right; highest design cost | +| **Single applier owner** | `recovery.Receiver` *or* `transport` replica handler | Whichever is canonical owns the `apply(lba, lsn, data)` path; the other becomes a thin demux/forward | +| **Replay source of truth** | Which encoding the on-disk WAL playback decoder reads | Determines whether existing on-disk WAL frames (V2-faithful) need a one-shot rewrite or a dual-decoder migration | + +### 10.2 What B-2 already buys us + +- Single live path: `Sender.PushLiveWrite` → `s.sink.NotifyAppend` is the only entry point. +- Sink interface stable: `StartSession / DrainBacklog / EndSession / NotifyAppend` — a real `transport.WalShipper` adapter satisfies this by duck typing without touching `recovery`. +- Architect P1 review rules 1+2 (emit context **before** `StartSession`; restore steady lineage **after** `EndSession`) are caller obligations that hold whether the sink is `senderBacklogSink` or the real adapter. +- Atomic-seal contract is owned by the sink, not `Sender` — when the bridging sink is replaced, the contract migrates with it. + +### 10.3 Pre-decision deliverables (parallel-safe) + +These do **not** require the format choice and can land while waiting on the decision: + +- **Adapter scaffolding**: `transport`-side struct that wraps `WalShipperFor(replicaID)` + lineage management; satisfies `recovery.WalShipperSink` by duck typing. Wire format remains TBD; tests use a recording `EmitFunc`. +- **Integration tests for architect rules 1+2**: assertions that the calling layer sets `(conn, lineage)` before `StartSession` and restores steady lineage after `EndSession`. Test the bridging sink today; carry over to the real adapter unchanged. +- **V2 port discipline note**: dual-lane v3 convergence first; V2 wire-compat is gated separately per `feedback_porting_discipline.md` (don't silently collapse incidental invariants). + +### 10.4 Ask, in one paragraph + +> P2c/B sink path is unified; bridging `senderBacklogSink` still duplicates `frameWALEntry` encoding because P2d is open. Going to a real `transport.WalShipper` sink requires **one** of: (A) unify dual-lane on `MsgShipEntry` payload + receiver dispatch, (B) teach `WalShipper.Emit` backlog path to encode `frameWALEntry`, or (C) a documented third (e.g. envelope byte). Plus: which package owns the **single applier** (`recovery.Receiver` vs `transport` replica handler), and which encoding is the **replay source of truth** for on-disk WAL. Pre-decision adapter scaffolding + rules-1+2 integration tests can land in parallel.