Documentation
¶
Overview ¶
Package postgres implements siphon's Postgres driver. Backup/Restore shell out to pg_dump/pg_restore for correctness; Inspect uses pgx for fast schema reads.
Index ¶
- type Conn
- func (c *Conn) ApplyChange(ctx context.Context, ch canonical.CanonicalChange) error
- func (c *Conn) Backup(ctx context.Context, opt driver.BackupOpts, w io.Writer) error
- func (c *Conn) BackupIncremental(ctx context.Context, since canonical.Position, w io.Writer) (canonical.Position, error)
- func (c *Conn) CaptureBaseEnd(ctx context.Context, info *IncrementalBaseInfo) error
- func (c *Conn) Close() error
- func (c *Conn) ConsumeCanonical(ctx context.Context, r io.Reader) error
- func (c *Conn) CreateBaseSlot(ctx context.Context) (*IncrementalBaseInfo, error)
- func (c *Conn) CurrentPosition(ctx context.Context) (canonical.Position, error)
- func (c *Conn) DropSlot(ctx context.Context, slotName string) error
- func (c *Conn) EmitCanonical(ctx context.Context, schema *canonical.CanonicalSchema, w io.Writer) error
- func (c *Conn) Inspect(ctx context.Context) (*driver.Schema, error)
- func (c *Conn) InspectSchema(ctx context.Context) (*canonical.CanonicalSchema, error)
- func (c *Conn) Restore(ctx context.Context, opt driver.RestoreOpts, r io.Reader) error
- func (c *Conn) StreamChanges(ctx context.Context, from canonical.Position, ...) (canonical.Position, error)
- func (c *Conn) SweepOrphanSlots(ctx context.Context) (int, error)
- func (c *Conn) Verify(ctx context.Context, r io.Reader) (*driver.VerifyReport, error)
- type Driver
- type IncrementalBaseInfo
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Conn ¶
type Conn struct {
// contains filtered or unexported fields
}
Conn is a live Postgres connection.
func (*Conn) ApplyChange ¶
ApplyChange applies one CanonicalChange to the database.
func (*Conn) Backup ¶
Backup spawns pg_dump and streams the output to w. ctx cancellation propagates to pg_dump via exec.CommandContext.
func (*Conn) BackupIncremental ¶
func (c *Conn) BackupIncremental(ctx context.Context, since canonical.Position, w io.Writer) (canonical.Position, error)
BackupIncremental captures the BOUNDED change set from `since` to the server's current end LSN, serializing each CanonicalChange to w as JSONL, and returns the end Position reached.
Bounding mechanism: the end LSN is captured up front via pg_current_wal_lsn() and passed to the shared pgoutput decode loop as a stop target. The loop advances its client position only AFTER decoding+emitting each XLogData message, so every change committed at or before the bound is emitted and none past it. The stop fires when either the decoded position reaches the bound OR a server keepalive reports the server's WAL end has reached it (catch-up: pgoutput delivers XLogData before the keepalive covering its LSN, so all changes <= the bound are already emitted, and the keepalive's position is tracked separately from the decoded position so it cannot truncate capture). This reuses StreamChanges' decode machinery rather than streaming raw WAL bytes, so the incremental body is engine-neutral JSONL that the restore path replays via ApplyChange.
Before streaming, orphaned siphon replication slots are swept (best-effort) to keep WAL retention bounded.
This path is exercised against a live wal_level=logical server only in CI (see incremental_integration_test.go); it is not validated locally (no Docker here).
func (*Conn) CaptureBaseEnd ¶
func (c *Conn) CaptureBaseEnd(ctx context.Context, info *IncrementalBaseInfo) error
CaptureBaseEnd records the end-of-base LSN into info.
func (*Conn) ConsumeCanonical ¶
ConsumeCanonical reads a stream produced by EmitCanonical and replays it into the database.
func (*Conn) CreateBaseSlot ¶
func (c *Conn) CreateBaseSlot(ctx context.Context) (*IncrementalBaseInfo, error)
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 (*Conn) CurrentPosition ¶
CurrentPosition records the resume anchor for the first incremental after a base backup. app.Backup calls this right after a full backup so the base dump's Envelope carries where the first incremental should resume from.
It also ESTABLISHES the logical replication slot if it does not yet exist. A logical slot only retains and decodes WAL produced after the slot's own creation, so the slot must exist from the recorded anchor forward or the first incremental silently captures nothing (it would resume from an LSN the slot never retained). When this call creates the slot, the slot's consistent point is the correct anchor — it is the exact LSN from which the slot guarantees decodable WAL. When the slot already exists it is already retaining WAL, so pg_current_wal_lsn() is a safe anchor.
func (*Conn) DropSlot ¶
DropSlot removes the replication slot once a chain is sealed (or via an orphan scan on startup). Safe to call best-effort.
func (*Conn) EmitCanonical ¶
func (c *Conn) EmitCanonical(ctx context.Context, schema *canonical.CanonicalSchema, w io.Writer) error
EmitCanonical writes a table-by-table snapshot of schema as JSONL to w.
func (*Conn) InspectSchema ¶
InspectSchema queries information_schema for tables (public schema) and their primary key columns, returning a CanonicalSchema.
func (*Conn) StreamChanges ¶
func (c *Conn) StreamChanges(ctx context.Context, from canonical.Position, emit func(canonical.CanonicalChange) error) (canonical.Position, error)
StreamChanges streams logical row changes as engine-neutral CanonicalChanges, starting after `from`. It uses pgoutput logical decoding over a dedicated replication-mode connection. Bounded callers cancel ctx at a target end position; unbounded (CDC) callers stream until ctx cancel. ctx cancellation is the normal stop signal and is NOT reported as an error — the final Position reached is returned for envelope stamping / CDC state persistence.
func (*Conn) SweepOrphanSlots ¶
SweepOrphanSlots drops inactive siphon-owned PHYSICAL base slots and returns the count dropped.
Policy: a base backup creates a non-temporary physical slot named siphon_<ulid> (see CreateBaseSlot) to pin WAL for a future incremental; that slot must be dropped when the chain is sealed. An active such slot is in use by a running backup, but an inactive one is orphaned — a normal run drops its own slot, so any inactive siphon_<ulid> left behind belongs to a crashed/aborted run and is only pinning WAL. We drop every inactive slot matching the prefix, EXCEPT the persistent logical CDC slot (siphonSlot), which is the resume anchor for change streaming and is legitimately inactive between runs — sweeping it would discard the resume position. Each drop is best-effort: a concurrent run that re-activates a slot between the scan and the drop makes pg_drop_replication_slot fail with "in use", which we skip.
type Driver ¶
type Driver struct{}
func (Driver) Capabilities ¶
func (Driver) Capabilities() driver.Capabilities
type IncrementalBaseInfo ¶
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
}
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).
Source Files