package granary

  1. Overview
  2. Docs
Legend:
Page
Library
Module
Module type
Parameter
Class
Class type
Source

Module Granary_replication.ReplicationSource

WAL-based physical replication to object store (Litestream-style).

This module defines the frame types carried on the replication stream, the asynchronous frame-sink interface that applications implement to ship frames to an object store, and the apply primitive that replays frames into local DB state for cold restore or a standby replica.

Sourcetype replicated_frame = {
  1. epoch : int64;
  2. frame_idx : int;
  3. page_id : int64;
  4. is_commit : bool;
  5. page : Cstruct.t;
  6. checksum : int64;
  7. source_salt : int64;
  8. source_seed : int64;
}

A single frame as it travels on the replication stream. The epoch identifies the WAL generation this frame belongs to; the consumer uses it to detect checkpoint/truncation boundaries and re-anchor.

checksum, source_salt, and source_seed store the transport-level checksum computed at the replication source. Callers can verify transport integrity with verify_checksum.

Sourcetype frame_sink = replicated_frame list -> unit Lwt.t

Application-implemented callback. Receives a committed batch of frames in order. The callback MUST be asynchronous and non-blocking so the commit path is never stalled.

Sourceval verify_checksum : replicated_frame -> bool

Recompute the frame checksum using source_salt and source_seed and compare against the stored checksum. Returns true if the frame was not corrupted in transit.

Sourceval apply_frames : wal:Granary_storage.Wal.t -> pager:Granary_storage.Pager.t -> replicated_frame list -> (unit, [> `Apply_error of string ]) result Lwt.t

Replay a batch of replicated frames into local DB state.

Each frame's page is written via Wal.append_commit, which recomputes the per-frame checksum using the local WAL's salt. Frame content integrity across the transport is verified independently by the caller (transport checksums).

  • Respects commit boundaries: frames up to and including the last is_commit frame form one append_commit batch. Trailing non-commit frames are silently ignored (the caller should ensure the input list ends at a commit boundary).
  • Grows the device via Pager.set_n_pages to accommodate the highest page_id in the batch (page ids are 0-indexed, so a frame for page N needs at least N+1 pages).
  • Returns Ok () when all committed batches are applied.

Note: this initial implementation does not inspect epoch for re-anchoring; it assumes all frames belong to a single WAL generation. Epoch-aware apply (with re-anchor from a fresh base snapshot) is deferred to the standby-apply driver in #172.

Sourceval cold_restore : read_at:(offset:int64 -> Cstruct.t -> (unit, string) result Lwt.t) -> write_at:(offset:int64 -> Cstruct.t -> (unit, string) result Lwt.t) -> sync:(unit -> (unit, string) result Lwt.t) -> wal_size_bytes:int64 -> pager:Granary_storage.Pager.t -> base_snapshot_path:string -> wal_frames:replicated_frame list Lwt_stream.t -> unit -> (unit, [> `Restore_error of string ]) result Lwt.t

One-shot cold restore: open a WAL over read_at/write_at/sync, then consume the wal_frames stream, accumulating frames into commit batches (up to each is_commit frame) and replaying them via apply_frames.

base_snapshot_path is informational only — the caller must have already loaded the base snapshot into the pager before calling this function. This module does not download or apply snapshots itself; it only replays WAL frames on top of the already-prepared pager state.

Sourceval incremental_restore : read_at:(offset:int64 -> Cstruct.t -> (unit, string) result Lwt.t) -> write_at:(offset:int64 -> Cstruct.t -> (unit, string) result Lwt.t) -> sync:(unit -> (unit, string) result Lwt.t) -> wal_size_bytes:int64 -> pager:Granary_storage.Pager.t -> incremental_sets:Granary_store.Store.backup_frame list list -> unit -> (int64 * int, [> `Restore_error of string ]) result Lwt.t

Incremental restore: chain a full base snapshot + ordered incremental frame sets captured via Granary_store.Store.capture_frames_since.

The caller must have already loaded the base snapshot into ~pager before calling this function — this module does not load snapshots; it only replays WAL frames on top of the already-prepared pager state.

Opens a WAL over the provided callbacks, then applies each incremental frame set in order via apply_frames_epoch_aware, handling epoch transitions transparently.

Returns the final (epoch, frame_idx) — the position of the last committed frame applied — or Ok (0L, -1) (the initial position) if no frames were provided. Returns Error `Restore_error on WAL-open failure or frame-apply failure.

--------------------------------------------------------------------

Standby apply driver (#172)

--------------------------------------------------------------------

Sourceval no_reader_gate : target:int -> unit Lwt.t

A no-op reader gate that returns immediately without waiting. Use for paths known to serve no concurrent readers (e.g. cold restore, incremental restore, or direct test invocations).

Sourceval checkpoint_wal_to_main : wal:Granary_storage.Wal.t -> pager:Granary_storage.Pager.t -> reader_gate:(target:int -> unit Lwt.t) -> (unit, [> `Apply_error of string ]) result Lwt.t

Checkpoint the standby's WAL into the main DB and reset it.

Migrates the latest version of every page in the WAL index to the main DB via Pager.flush_one_to_main, syncs, then calls Wal.reset (which bumps the WAL Granary_storage.Wal.epoch). This is the standby-side equivalent of the engine's inline checkpoint (Granary_store.Store.checkpoint) and is the primitive used to drain buffered committed frames to durable storage before the WAL is recycled — both on master-epoch transitions and on promotion.

Sourceval apply_frames_epoch_aware : wal:Granary_storage.Wal.t -> pager:Granary_storage.Pager.t -> last_epoch:int64 -> last_idx:int -> reader_gate:(target:int -> unit Lwt.t) -> replicated_frame list -> (int64 * int, [> `Apply_error of string ]) result Lwt.t

Apply a batch of replicated frames into the standby's WAL and pager, handling epoch transitions.

Behaves like apply_frames except that when the incoming frame epoch differs from last_epoch, the standby's local WAL is checkpointed (migrated to the main DB and reset) before the new epoch's frames are applied. This keeps the standby's WAL frame indices from colliding with recycled indices from the master's checkpoint cycle.

Returns Ok (epoch, frame_idx) of the last applied commit frame, or Ok (last_epoch, last_idx) if no new frames were committed (useful for lag-tracking). Pass ~last_epoch:0L ~last_idx:(-1) for the initial call.

A single batch may safely bundle frames from more than one master epoch (#209): the batch is split into maximal single-epoch runs and each run is fed through the epoch-transition logic in order, so an intervening checkpoint between epoch N and N+1 frames is always honored before the N+1 frames are appended.