Boot recovery of wal: :dual (ADR 0002, section 8), and the same code offline as mix ankusa.recover.

A replacement node starts with the same intake.node_id as the lost one, restores from the backup (Ankusa.Store.Backup), then run/2:

1. Identity. m:intake_node is stamped on first use; a different id refuses ({:intake_node_mismatch, stamped, configured}): restoring node a's backup into node b would interleave two seq spaces.
2. Topology. Ankusa.Intake.verify/1 (skipped with intake.verify_topic: false); the node waits until the log answers.
3. Replay the tail from m:intake_offset + 1 (:earliest when the store has no offset), in chunks of 500 in one synced batch each, the offset advancing in the same batch. A record is applied through one rule, keyed by seq and hook id: no row at hooks/<seq> writes the hook under its original seq (delivery rows from the source's current config, dedupe keys, counters, m:next_seq raised); the same id is skipped; a different id is a seq collision, written under a fresh seq, never over the other hook. A duplicate, reordered or twice-replayed record changes nothing.
4. How far back. If the log's earliest position is above offset + 1 and the store was restored, {:intake_truncated, wanted, earliest} refuses (allow_gap: true goes on and says so); on an existing store it is a warning, because every hook in the gap is in the local store.
5. Republish the local tail: every hooks row above m:intake_seq (a crash can leave local hooks the log never confirmed), then the watermark moves to next_seq - 1.

It runs as an Ankusa.Instance child after the source store (replay resolves sinks through it) and before the writer: the node serves nothing until it is done, and Health.ready/1 says intake: "failing" (reason recovering) meanwhile.