Once a primary database has written a record to its own write-ahead log, PHYSICAL REPLICATION streams that record to one or more standbys, which apply it to keep their own copy of the data caught up. This is what Postgres's walsender.c (primary side) and walreceiver.c (standby side) do: the primary sends WAL records starting from wherever a standby asks to resume, and the standby applies whatever new records show up, in order.

This is a different problem from this course's WAL Crash Recovery exercise, not a re-skin of it: crash recovery replays a FIXED, already-complete log exactly once, filtering out any transaction that never reached COMMIT. Here, records arrive over a live, resumable STREAM -- and by the time a record leaves the primary's WAL it has already been decided, so there's no commit-filtering to do. The hard part instead is "have I already applied this" -- a standby that disconnects and reconnects will typically have its last few records RESENT (the primary can't be sure what made it across before the drop), and must apply each one EXACTLY once, in LSN order, without losing anything or replaying from the very start.

Each WAL record is a dict: {"lsn": N, "op": "SET"/"DELETE", "key": K, "value": V} ("value" absent for DELETE). LSN (log sequence number) is a byte offset into the log, not a counter -- it strictly increases, but NOT necessarily by 1 each time, so never assume consecutive integers or that every requested resume point is an exact match for some record's lsn.

_apply_record (given) applies one record's effect to a state dict, with no LSN bookkeeping. Implement:

• fetch_since(primary_log, since_lsn): the primary/sender side -- given its full WAL and a standby's requested resume point, return every record with lsn > since_lsn, in ascending LSN order.
• Standby.apply_batch(records): the standby/receiver side -- given a batch of records, apply each genuinely new one to self.state via _apply_record and advance self.last_applied_lsn, but SKIP any record whose lsn is already <= self.last_applied_lsn at the moment it's reached (a resend from before the last disconnect) rather than re-applying it or raising.

Example

standby = Standby()
standby.apply_batch(fetch_since(primary_log, standby.last_applied_lsn))
# ... connection drops, standby reconnects later ...
standby.apply_batch(fetch_since(primary_log, standby.last_applied_lsn))
# -> resumes exactly where it left off; any overlap in what gets resent
#    is silently absorbed, nothing is applied twice or skipped

Constraints

• fetch_since must return records sorted ascending by lsn, regardless of what order they appear in primary_log.
• apply_batch must tolerate a batch containing records the standby has already applied (by lsn) mixed in with genuinely new ones -- already -applied records are silently skipped, not reapplied and not treated as an error.
• self.last_applied_lsn must only ever move forward, and only ever to the lsn of a record that was actually applied.

Real-world grounding

• Postgres's WAL sender (primary side of streaming replication): src/backend/replication/walsender.c
• Postgres's WAL receiver (standby side): src/backend/replication/walreceiver.c

Real streaming replication adds a lot more: the standby periodically reports back its own flushed/applied LSN so the primary knows it's safe to recycle old WAL segments, synchronous replication modes that block a primary's commit until a standby confirms receipt, and WAL retention limits (if a standby falls far enough behind that the primary has already discarded the segment it needs, replication breaks and requires a full resync, not just a resumed stream) -- but "resume from the requested LSN, apply exactly once, tolerate resent records" is the core contract underneath all of it.