pgnotch

README

pgnotch

Append-only, fenced logs in stock PostgreSQL. No extension, no background worker, no server-side code — a table, a row, and one statement per append.

go get github.com/aromanovich/pgnotch

A log is a gap-free sequence of entries under an identifier you choose. One writer owns it at a time, and ownership is an epoch: fencing a log at an epoch cuts off every append of a lower one, so a writer that has lost the log finds out at its next append instead of writing over its successor. The owner assigns sequence numbers itself, which is what lets an append be a single round trip and the order be total with nothing coordinating it.

// Every name this package writes is unqualified, so it lands in the schema the
// connection's search_path names. That is the whole of how one deployment's
// logs are kept apart from another's in the same database.
pool, err := pgxpool.New(ctx, "postgres://…/db?search_path=orders")
if err != nil {
    return err
}
defer pool.Close()

if err := pgnotch.Migrate(ctx, pool); err != nil {
    return err
}
store, err := pgnotch.Open(ctx, pool)
if err != nil {
    return err
}

const shipments = pgnotch.LogID("shipments")

// Create the log. This is the only call that ever makes a table, and it is
// idempotent — run it over your whole set of logs at start-up. A log you
// never created cannot be fenced: you get ErrNoSuchLog.
if err := store.CreateLogs(ctx, shipments); err != nil {
    return err
}

// Claim it. The epoch comes from whatever hands out ownership — a lease
// counter, a ZooKeeper czxid, a database sequence — and must be strictly
// greater on each new claim.
if err := store.Fence(ctx, shipments, epoch); err != nil {
    return err
}

// Ask where the next append goes: the log you have just fenced may be one
// somebody else was writing, and this is what hands that over.
next, err := store.NextSeqno(ctx, shipments)
if err != nil {
    return err
}

// Append at consecutive seqnos. Returning nil means every entry up to the
// batch's last is durable.
err = store.Append(ctx, shipments, epoch, next, [][]byte{
    []byte("first"),
    []byte("second"),
})
switch {
case errors.Is(err, pgnotch.ErrFenced):
    return fmt.Errorf("the log is somebody else's now: %w", err)
case errors.Is(err, pgnotch.ErrAlreadyWritten):
    // The batch is already there — this is the ack for a previous attempt
    // that failed ambiguously, not a failure.
case err != nil:
    return err
}

// The next batch starts where this one ended: seqnos are always the caller's
// to assign, and there is no "append at the end". The owner keeps its own
// high-water mark from here and does not ask again; the three outcomes above
// are this call's as well.
err = store.Append(ctx, shipments, epoch, next+2, [][]byte{
    []byte("third"),
})

entries, err := store.ReadFrom(ctx, shipments, pgnotch.FirstSeqno, 100)

A writer that has just fenced a log somebody else wrote does not have that mark, and NextSeqno hands over where the next append goes: one registry row by primary key. It is kept there rather than derived from the entries, so it is right for a log a trim has emptied — where reading the log for it would find nothing and start again at the first seqno. Ask once on taking the log and track it from there; an append does not need to ask. A replay after an ambiguous append is the other case that looks like a new seqno and is not — resend the same batch at the same seqno and read ErrAlreadyWritten as the ack.

Requires PostgreSQL 16 or newer (bytea STORAGE PLAIN in CREATE TABLE arrived there) and pgx/v5.

What you can rely on

1. Total order per log. The owner assigns seqnos, so there is no tie to break and no clock in the design.
2. Fencing. A completed Fence cuts off every append of a lower epoch, atomically.
3. Cumulative ack. An Append that returns nil means every entry up to and including the batch's last seqno is durable, so "confirmed ⟺ seqno ≤ the last acked" is inherited rather than implemented.
4. Gap-freedom. An append never skips a seqno, so a log is one unbroken run: Trim moves its lower end, an append its upper end, and nothing puts a hole in the middle.
5. Readback. ReadFrom returns every entry a completed append acked and no trim has removed, in seqno order — across a change of owner included.
6. Handover. NextSeqno names the seqno a new owner's first append must start at — for a log whose entries a trim has all taken included.

Payloads are opaque bytes. Nothing here interprets, compresses or frames them, and nothing here decides what a log is for.

The three refusals

Append has exactly three outcomes a caller is expected to handle, none of which writes anything:

error	means	what to do
ErrFenced	the log belongs to another epoch, or to nobody	stop writing; you are not the owner
ErrAlreadyWritten	a seqno in the batch is taken	if this is a replay of the same batch, it is the ack
ErrGap	the entry below the batch is missing	retry once the predecessor lands

Where more than one applies, ErrFenced wins. ErrAlreadyWritten is an ack, and a writer that has already lost its log would otherwise take its successor's word for its own high-water mark.

That ranking is what makes the retry rule safe: after an append that failed ambiguously — a dropped connection, a cancelled context — replay the same batch and read ErrAlreadyWritten as "it landed". A batch that overlaps the log only partly is refused whole and is a caller bug, and a writer whose epoch grew across the ambiguity must replay under the epoch it holds now.

How it works

The registry row decides everything. Per log there is one row holding the owning epoch, the last seqno appended and the trim watermark. An append is one UPDATE whose WHERE clause is simultaneously the fencing check, the gap check and the already-written check, so all three are decided atomically under that row's lock. Nothing is derived from the entry rows — which is what lets a trim take all of them while ErrAlreadyWritten still answers for the seqnos it removed.

Two appends to one log serialise on that row, and PostgreSQL picks the winner. The check is the UPDATE's own WHERE, so the lock is the one that UPDATE takes and no SELECT … FOR UPDATE is issued anywhere. The second append waits on it, and under READ COMMITTED an UPDATE released from that wait re-reads the row it waited for and applies the predicate again: if the winner committed, last_seqno has moved and the loser matches nothing, which is its ErrAlreadyWritten; if the winner rolled back, the predicate holds again and the loser proceeds as though it had been alone. Uniqueness is therefore a property of that one row, which is why the entry tables can carry no index and still not admit a seqno twice.

An append is one statement and one round trip. The UPDATE and the entry rows are a single CTE, not a transaction around two. A transaction would spend a round trip on BEGIN, one on the UPDATE, one on the rows and one on COMMIT, and hold the row every writer of that log contends for across all four.

The entry tables carry no index and no TOAST. The writer assigns seqnos, so there is nothing to look up by identity — and a unique index would be worse than useless: _bt_check_unique reads under SnapshotDirty, and rows a trim has removed are not there to be seen, so a re-append into the trimmed prefix would pass it. The payload column is bytea STORAGE PLAIN, which keeps PostgreSQL from ever building a TOAST relation and the btree under it; entries larger than a page are chunked here instead and reassembled on read, so a caller never sees a chunk.

Full-page images cost a constant per checkpoint, not a rate per entry. An append-only table only ever touches the page it is filling, so after a checkpoint has passed under it exactly that page and its free-space map page carry an image and every further row carries none.

Space comes back by TRUNCATE, never by DELETE. Each log has two entry tables used as a ring, and the half a trim has emptied is truncated — which discards its dirty buffers without writing them, resets its freeze horizon and takes its whole vacuum debt with it. A generation that dies before the next checkpoint never reaches the disk at all.

Operating it

Migrations are goose, and they are plain SQL files. They live in migrations/ and are embedded, so Migrate applies them with nothing installed; Open refuses a schema nobody has migrated (ErrNotMigrated) rather than conjuring tables from under you.

Nothing in the files is this package's to supply, so the goose CLI applies the very same directory against the very same DSN:

goose -dir migrations -table pgnotch_migrations \
      postgres "postgres://…/db?search_path=orders" up

-table is what keeps this schema's version out of goose_db_version, so your own migrations can share a schema with these without the two disagreeing about what version it is at. status, up-to and down work the same way. In process, Provider hands back the goose provider itself:

db := stdlib.OpenDBFromPool(pool) // closing this leaves the pool open
defer db.Close()

provider, err := pgnotch.Provider(db)
status, err := provider.Status(ctx)

The down is not a gentler Drop. There is one migration and it is the registry, which is also the only thing that can enumerate the entry tables — so rolling it back has to take every log's entries with it, or strand them where nothing could ever name them again. goose down destroys exactly what Drop does, and leaves goose's version table behind. If you meant "roll back a schema change", there is not one yet; if you meant "these logs are finished", use Drop — or drop the schema.

Half the schema is not versioned, and cannot be. There is one pair of entry tables per log and you may create logs forever, so they are created by CreateLogs, inside the transaction that registers each log — a registry row always has its tables. The cost: a schema change to the entry tables would be a migration against an unbounded set of tables, and there is no such migration today.

Size the driver's statement cache for your logs. An append names its own log's tables, so its statement text is that log's alone and a connection holds one prepared statement per log it has touched. pgx caches 512 by default and evicts by LRU: a connection round-robining across more logs than that misses on every append, and the single round trip becomes three. It is fixed in the DSN, not in this package:

postgres://…/db?statement_cache_capacity=2048

At 1024 logs the difference measured three round trips and 1985 bytes an append against one and 1151 with the cache sized for the log count.

Leave the isolation level at READ COMMITTED. The three refusals are decided by a predicate that is re-evaluated after a wait, and that re-evaluation is READ COMMITTED's alone. Point this package at a DSN or a server set to repeatable read and a contended append comes back SQLSTATE 40001, could not serialize access due to concurrent update — an error this package wraps and passes on, not one of the three — so the writer that would have read ErrAlreadyWritten as its ack is told something it has no rule for. Nothing here sets the level, and nothing here can tell that it was changed.

A schema is the unit of separation, and it is PostgreSQL's own. Every name this package writes is unqualified — pgnotch_logs, pgnotch_entries_<n>_<slot>, pgnotch_migrations — so all of them land wherever the connection's search_path points. Give a deployment a schema and it shares nothing with the next one:

postgres://…/db?search_path=orders

Two Stores over one schema are two writers of the same logs, which is what a failover looks like from here. This package neither creates the schema nor assumes it is empty.

Drop removes the tables this package owns and leaves the schema alone. It finds the per-log entry tables through the registry rather than by matching a name pattern, because there is no bound on how many there are and the registry is the only complete list. An operator giving the whole thing back wants DROP SCHEMA … CASCADE instead.

Create your logs; nothing here creates one for you. A log is two tables, an id is an arbitrary string you supply, and PostgreSQL does not reclaim a table because nobody wanted it. If a fence conjured a log, one bad id — a wrong tenant, an unescaped input, a retry loop with a counter in it — would leave tables behind at whatever rate you called it, and a million of them is a pg_class you cannot clean up after the fact. So CreateLogs is explicit and Fence answers ErrNoSuchLog:

// Idempotent, so run it over the whole set on every start.
err := store.CreateLogs(ctx, ids...)

Over a set that is already there it does no DDL at all: the insert returns the rows it actually created, which on a restart is none. Creating a set that is new costs one statement of DDL for the whole batch rather than two per log, so provisioning a shard range at start-up is a handful of round trips and not thousands. Where your id space is genuinely bounded and you know the bound — a fixed shard count, a tenant list — this is the shape to reach for.

Trim, or pay for every entry ever written. Nothing here trims on its own: the log does not know which entries you have finished with. Trim moves the watermark synchronously and reclaims the space when it can — reclamation takes an ACCESS EXCLUSIVE lock with a 250 ms timeout, so a backup holding the table off delays the space and never the log.

What this is not

• not a replicated log. Durability and availability are PostgreSQL's, which means whatever your replication and failover give you and nothing more. There is no quorum here and no leader election — fencing tells you when you have lost a log, it does not decide who gets it;
• not a queue or a broker. There are no consumer groups, no acks per reader, no delivery semantics. A reader tracks its own position and reads from it;
• not multi-writer. One epoch owns a log. Concurrent appends to one log from two holders of the same epoch race for seqnos and lose, and fencing cannot separate them — whoever hands epochs out owes a strictly greater one per acquire;
• not a place for large values by default. Entries above a page are chunked and reassembled, which works at any size PostgreSQL can hold, but a megabyte entry is a megabyte through the connection on every read of it.

Tests

The suite needs a PostgreSQL 16 or newer and is pointed at it by environment variable. It creates a schema of its own per test and drops it afterwards.

PGNOTCH_DSN=postgres://user:password@localhost:5432/db go test ./...

make pg-up starts one in a container for exactly this — a tmpfs data directory, since nothing a test writes is meant to outlive the run — and make test points the suite at it; make pg-down takes it away again.

Without PGNOTCH_DSN the suite does not skip — it fails. There is no configuration in which go test ./... is green having never spoken to PostgreSQL, and that is deliberate: this package is a claim about what PostgreSQL does, so a run that stayed inside the process is not a weaker result, it is a different one wearing the same colour.

rules_test.go holds the rules that are pure functions of values — the ranking between two refusals that both apply, the chunk boundary either side, the ring's arithmetic at zero. A driver test reaches one path of each per run, and some of them only by racing two writers, so they are stated there as tables. They need no database and they run under the same rule anyway: they are the cases the driver tests cannot reach, not a suite of their own.

Two of them are cost guards rather than correctness tests — one for what an append writes, one for what it waits for — and both were written by breaking the implementation on purpose and watching them go red. They are the reason this package is worth having over any other way of putting a log in a table, so a change that makes either of them fail is a change to what this package is.

Putting load on it

cmd/pgnotch-load is a load generator: it takes everything from its flags, appends at the rate they name until it is interrupted, and prints what the window did.

make load ARGS='-rps 500 -logs 8 -sizes 1k:9,32k:1'
# or, pointed anywhere yourself
go run ./cmd/pgnotch-load -dsn postgres://user:password@localhost:5432/db -rps 500

8 logs at epoch 1787607891, 500 appends/s × 1 entries = 500 entries/s, ~2.0 MiB/s
sizes 1.0 KiB×9 32.0 KiB×1, mean 4.1 KiB, 10% over the 8000-byte chunk
each log keeps 100000 entries, trimmed every 12500
[    5s]     2500 appends    500.0/s |     2500 entries |  10.1 MiB   2.0 MiB/s |     3480 rows    250 cut | p50  319µs p99  639µs max  1.5ms

The distribution is the point of the tool. -sizes takes size:weight classes — 1k:9,32k:1 is one entry in ten at 32 KiB — and a class over MaxEntryChunk is an entry the library has to cut into several rows, which is a different write from a 900-byte one and is not exercised by a single size. rows and cut on the report line are how much of that the window actually did.

flag
-rps	appends a second over all logs together
-logs	logs, which is also the number of concurrent writers: one log admits one
-batch	entries per append
-sizes	the payload-size distribution, size:weight with k and m suffixes
-retain	entries a log keeps before the writer trims behind itself; 0 never trims
-duration	how long to run; 0 runs until SIGINT
-schema	the schema to put the logs in, created if missing

Three things follow from what a log is, rather than from the tool:

• it owns the logs it writes. They are created under -prefix, fenced at an epoch of the run's own — Unix seconds unless you pass -epoch — and a run that finds one taken by a higher epoch stops with ErrFenced rather than fencing it back, because a second generator racing the first would be measuring the race;
• it trims behind itself, which is what makes an unbounded run cost a bounded amount of disk. -retain 0 turns that off and the tables then grow for as long as it runs;
• it picks up where it left off. A restart asks each log where its next append goes, the way any new owner has to, and says which seqno it continues at.

The rate is a schedule fixed when the run starts, not a sleep between appends, so a slow round trip is repaid out of the slots after it rather than lowering the rate quietly. When it cannot be repaid the slots are abandoned and counted as skipped: a run that could not keep the rate you asked for says so on the line, and so does a run whose appends are failing — the count and the last error, since a load generator writing nothing looks exactly like one writing everything.

License

MIT — see LICENSE.

Documentation

Overview

Package pgnotch keeps append-only, fenced logs in stock PostgreSQL: no extension, no background worker, no server-side code of its own.

A log is a gap-free sequence of entries under an identifier the caller chooses, created by Store.CreateLogs and never as a side effect. One writer owns it at a time, at an epoch Store.Fence takes, so a writer that has lost its log finds out at its next append. The owner assigns seqnos itself, which lets an append be a single statement.

What a caller may rely on

1. Total order per log, with no tie to break and no clock in the design.
2. Fencing. A completed Store.Fence cuts off every append of a lower epoch, atomically.
3. Cumulative ack. An Store.Append that returns nil means every entry up to and including the batch's last seqno is durable.
4. Gap-freedom. An append never skips a seqno: Store.Trim moves a log's lower end, an append its upper end, and nothing puts a hole between.
5. Readback. Store.ReadFrom returns every entry a completed append acked and no trim has removed, in seqno order, across a change of owner.
6. Handover. Store.NextSeqno names the seqno a new owner's first append must start at, for a log whose entries a trim has all taken included.

Payloads are opaque bytes; nothing here interprets or frames them.

How it works

Per log there is one row in the registry table holding the owning epoch, the last seqno appended and the trim watermark. An append is one statement: an UPDATE whose WHERE clause is at once the fencing check, the gap check and the already-written check, decided under that row's lock, with the entry rows in the same CTE. Nothing is derived from the entry rows, so a trim can take all of them while ErrAlreadyWritten still answers for their seqnos.

The entry tables carry no index: the registry row is the authority on which seqnos are spent, and `_bt_check_unique` reads under SnapshotDirty, so an index could not see the rows a trim removed. Nor a fillfactor — measured, it only moved the full-page image to the free-space map page, at eight times the space for a 900-byte entry. The payload column is `bytea STORAGE PLAIN`, which keeps PostgreSQL from ever creating a TOAST relation for these tables, at the price of a hard "row is too big" past a page, so entries are chunked at MaxEntryChunk and a caller never sees a chunk.

Space comes back by TRUNCATE, never by DELETE: each log has two entry tables used as a ring, and the one a trim has emptied is truncated, discarding its dirty buffers unwritten and taking its whole vacuum debt with it.

Operating it

An append is one round trip only while the driver has the statement prepared. It names its own log's tables, so a connection holds one per log it has touched; a smaller driver cache misses every append and the round trip becomes three. pgx caches 512, set in the DSN by `statement_cache_capacity`.

Migrate applies the static half of the schema and Open refuses one nobody has migrated. The migrations are ordinary goose SQL files, embedded here and shipped in `migrations/`; Provider hands back the goose provider itself, with the warning it carries about the down. The per-log entry tables are not versioned and cannot be — Store.CreateLogs makes them in the transaction that registers each log, and a migration over an unbounded set of tables is not something this package has.

Every name this package writes is unqualified, so all of them land in the schema the connection's search_path names: two [Store]s over one schema are two writers of the same logs, two over different schemas share nothing.

Requires PostgreSQL 16 or newer: `bytea STORAGE PLAIN` in CREATE TABLE, rather than a following ALTER, arrived there.

Index

• Constants
• Variables
• func Drop(ctx context.Context, pool *pgxpool.Pool) error
• func Migrate(ctx context.Context, pool *pgxpool.Pool) error
• func Provider(db *sql.DB, opts ...goose.ProviderOption) (*goose.Provider, error)
• type Entry
• type Epoch
• type LogID
• type Seqno
• type Store
  • func Open(ctx context.Context, pool *pgxpool.Pool) (*Store, error)
  • func (s *Store) Append(ctx context.Context, id LogID, epoch Epoch, first Seqno, payloads [][]byte) error
  • func (s *Store) CreateLogs(ctx context.Context, ids ...LogID) (err error)
  • func (s *Store) Fence(ctx context.Context, id LogID, epoch Epoch) error
  • func (s *Store) NextSeqno(ctx context.Context, id LogID) (Seqno, error)
  • func (s *Store) ReadFrom(ctx context.Context, id LogID, from Seqno, limit int) ([]Entry, error)
  • func (s *Store) Trim(ctx context.Context, id LogID, upTo Seqno) error

Constants

const MaxEntryChunk = 8000

MaxEntryChunk is the payload one row carries: a heap tuple must fit a page (PostgreSQL's MaxHeapTupleSize is 8160), the row's other columns and headers take about fifty bytes, and the rest is slack. A larger entry becomes several rows, which a caller never sees — entries go in whole and come back whole.

const MaxLogIDBytes = 255

MaxLogIDBytes bounds a LogID, so an id too long is refused where it is passed rather than at the first append that overflows an index entry.

Variables

var (
	// ErrFenced means the log is not the caller's to write: some other epoch
	// has fenced it, or the caller never fenced it at its own epoch. The caller
	// must stop writing.
	ErrFenced = errors.New("pgnotch: log is not fenced at this epoch")

	// ErrAlreadyWritten means a seqno the append asked for is taken. After an
	// append that failed ambiguously it says the write landed, and so is a
	// retry's success signal — provided the retry is the same batch: same first
	// seqno, same number of payloads, nothing added to it. A batch overlapping
	// the log only partly is refused whole. [ErrFenced] outranks it, so a
	// writer whose epoch grew across the ambiguity must replay under the epoch
	// it holds now, or read the log, to learn whether the first attempt landed.
	ErrAlreadyWritten = errors.New("pgnotch: seqno already written")

	// ErrGap means the append would leave a hole: the entry below the batch is
	// missing. Expected of a pipelined append that arrived out of order; retry
	// once the predecessor lands.
	ErrGap = errors.New("pgnotch: predecessor seqno is missing")

	// ErrNoSuchLog means the log has not been created. [Store.Fence] returns it
	// rather than creating one: see [Store.CreateLogs].
	ErrNoSuchLog = errors.New("pgnotch: no such log")

	// ErrZeroEpoch is what a caller that forgot to set an epoch gets, rather
	// than an [ErrFenced] that reads like a lost log: epoch 0 means "nobody
	// owns this".
	ErrZeroEpoch = errors.New("pgnotch: epoch 0 is not a valid epoch")

	// ErrNotMigrated means the schema has no tables of this package in it yet.
	// [Open] returns it rather than creating tables, so a process which is not
	// the one that deploys cannot become the one that migrates; call [Migrate]
	// and open again.
	ErrNotMigrated = errors.New("pgnotch: schema is not migrated")
)

Errors a caller is expected to handle; anything else it can only report. Match with errors.Is; the errors returned wrap these with context.

Functions

func Drop

func Drop(ctx context.Context, pool *pgxpool.Pool) error

Drop removes every table this package owns: its logs' entry tables, the registry and goose's version table. The schema itself is left alone; an operator giving that back too wants DROP SCHEMA.

A Store does not survive it: it caches which tables each log's entries are in, and since ordinal is GENERATED ALWAYS AS IDENTITY the identity goes with the registry table, so a re-migrated schema hands out ordinal 1 again and those cached names are either gone or a later log's, appended to without error. Drop what nothing is using, and open again afterwards.

func Migrate

func Migrate(ctx context.Context, pool *pgxpool.Pool) error

Migrate applies the schema to whatever schema the pool's search_path names. It is safe to run concurrently with itself and with a running Store: goose takes the version table's lock, and no migration touches a log's entry tables. It does not close pool.

func Provider

func Provider(db *sql.DB, opts ...goose.ProviderOption) (*goose.Provider, error)

Provider is the goose provider for this package's schema, for an operator who wants what goose offers beyond Migrate: status, a targeted up-to, a down. The down takes every log's entries with it, since the registry migration is the only thing that can enumerate the entry tables.

db may be anything that speaks to the right database; stdlib.OpenDBFromPool turns a pool into one and closing the result leaves the pool open. Options given here are applied after this package's own, so goose.WithLogger gets back the logging the goose.NopLogger installed here takes away.

Types

type Entry

type Entry struct {
	// Seqno is the entry's position in its log.
	Seqno Seqno
	// Epoch is the epoch its writer held when it appended the entry. Epochs
	// are non-decreasing along a log.
	Epoch Epoch
	// Payload is the bytes the caller appended, and is never nil for an entry
	// a read returns: an empty payload comes back empty, not missing. The
	// bytes alias neither this package's state nor another entry of the same
	// read, so the caller may keep them and decode into them in place.
	Payload []byte
}

Entry is one record of a log.

type Epoch

type Epoch uint64

Epoch is the ownership token every append carries. It may grow without an ownership change: a writer renewing its claim fences again at a higher epoch and keeps its log.

Zero is not a valid epoch; see ErrZeroEpoch. Whoever hands epochs out owes this package a strictly greater epoch per acquire, since fencing cannot separate two writers holding the same one.

type LogID

type LogID string

LogID names one log: an opaque key this package never interprets. Logs under different ids share nothing.

It is stored as `text` and never appears in an identifier — a log's tables are named from an ordinal assigned at creation — so the constraints are only that it be at most MaxLogIDBytes, not empty, valid UTF-8 and free of NUL.

type Seqno

type Seqno uint64

Seqno is the position of an entry in one log: an LSN the owner assigns itself. Total order within a log, no gaps.

const FirstSeqno Seqno = 1

FirstSeqno is the seqno of a log's first entry. Seqnos below it are not entries; they are reserved for this package's own bookkeeping.

type Store

type Store struct {
	// contains filtered or unexported fields
}

Store is the logs in one PostgreSQL schema, whichever the pool's search_path names. It is safe for concurrent use, but two concurrent appends to the same log race for seqnos and lose. It does not own the pool and never closes it.

func Open

func Open(ctx context.Context, pool *pgxpool.Pool) (*Store, error)

Open returns a Store over the schema the pool's search_path names, which must already have been migrated: see Migrate, and ErrNotMigrated for why this is not done here. It does not take ownership of pool.

func (*Store) Append

func (s *Store) Append(ctx context.Context, id LogID, epoch Epoch, first Seqno, payloads [][]byte) error

Append writes payloads as entries at consecutive seqnos starting at first, under epoch, as one atomic unit. first must be at least FirstSeqno and the batch must hold at least one payload; an individual payload may be empty. The payloads stay the caller's: no slice is retained or read after Append returns. Returning nil means the batch is durable through its last seqno.

Errors:

• ErrFenced when the log belongs to another epoch, or to nobody,
• ErrAlreadyWritten when any seqno in the batch is taken,
• ErrGap when the entry below first is missing.

None of the three writes anything; where more than one applies ErrFenced wins, because ErrAlreadyWritten is an ack a lost writer would trust.

It is one statement: an explicit transaction would hold the registry row across BEGIN, the UPDATE, the rows and COMMIT. COPY, the only path to heap_multi_insert, carries neither the predicate nor the refusal, so the rows go in by INSERT and pay a WAL record apiece rather than one per page — 41 bytes a row more at 900-byte entries (~4%), nothing at page-sized ones, where a row is its own page.

func (*Store) CreateLogs

func (s *Store) CreateLogs(ctx context.Context, ids ...LogID) (err error)

CreateLogs brings logs into existence, and is the only thing here that ever creates a table. It is idempotent: ids that already exist are left exactly as they are, ownership and entries included. A created log has no owner, so an Store.Append to it is refused with ErrFenced until someone fences it, and either every id in the batch exists when this returns nil, or none of the ones it had to create do.

Creation is separate from fencing because a LogID is an arbitrary string: a fence that conjured a log would leave tables behind on one bad id.

func (*Store) Fence

func (s *Store) Fence(ctx context.Context, id LogID, epoch Epoch) error

Fence claims a log for epoch, atomically cutting off every append of a lower epoch, so a writer that has lost the log cannot slip an append past a completed Fence. It is idempotent per epoch, so a restart without a change of ownership can replay the same acquire path; fencing at a higher epoch is how the same owner renews, and at a lower one fails with ErrFenced.

Ownership is all a fence changes: the entries stay and the new owner continues the log at the next seqno. A failed fence changes nothing, and fencing a log that does not exist is ErrNoSuchLog and creates nothing.

func (*Store) NextSeqno

func (s *Store) NextSeqno(ctx context.Context, id LogID) (Seqno, error)

NextSeqno is the seqno the log's next append must start at: one past its last entry, and FirstSeqno for a log nothing has appended to. A log that does not exist is ErrNoSuchLog. It is the registry row's and is not derived from the entries, so it is right for a log a trim has emptied.

A new owner asks once and keeps the mark itself from there: the log's end after an Store.Append is that batch's last seqno, and asking again per append would put a second round trip on the one call that has only one.

It answers for the log only while the caller owns it — an append by a higher epoch moves it — which is not a race to lose: appending at a stale value is refused rather than misplaced.

func (*Store) ReadFrom

func (s *Store) ReadFrom(ctx context.Context, id LogID, from Seqno, limit int) ([]Entry, error)

ReadFrom returns up to limit entries of the log with seqno at or above from, in seqno order. limit must be positive, and a from below FirstSeqno reads from FirstSeqno. Fewer than limit entries means the log ends there, so a caller reading a whole log loops until a short read.

A log nothing has fenced reads as empty, and a read after a successful Store.Fence sees every entry the log held when the fence took it.

func (*Store) Trim

func (s *Store) Trim(ctx context.Context, id LogID, upTo Seqno) error

Trim removes the log's entries at or below upTo. Trimming entries that are not there is not an error: Trim states where the log should start, and repeating it is harmless. The log stays appendable at the next seqno, ownership stays put, and the seqnos removed stay spent — an append at one is ErrAlreadyWritten.

The watermark moves synchronously and the space comes back when it can: a read is gated on the watermark, so rows may outlive it, never their visibility.