Documentation
¶
Overview ¶
Package streamclient is the reference implementation of ghsync's public Postgres change-stream contract. It owns watermark-bounded paging, durable cursors, transactional handler delivery, bootstrap, and RESYNC_REQUIRED.
ghsync v1 deliberately binds this library to pgx v5 and database co-location. External services consume the stream through this library and a shared Postgres database connection; v1 does not provide a wire API.
Index ¶
Constants ¶
This section is empty.
Variables ¶
ErrListenerUnavailable classifies a transient failure to acquire or use the dedicated LISTEN connection, including connection-pool exhaustion. Tail handles this condition internally by polling and reconnecting, so callers normally observe it only through an error wrapped by their own instrumentation.
Functions ¶
func IsRetryable ¶
IsRetryable reports whether an operation may be attempted again without a Bootstrap. Cursor contention, transient listener/pool unavailability, PostgreSQL serialization/deadlock/connection-capacity failures, and pgx failures known to occur before a request was sent are retryable.
Context cancellation and deadline expiry are not retryable with the same context. ErrResyncRequired is also false: it requires Bootstrap and a projection replacement before Tail resumes.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client consumes ghsync change streams from the same Postgres database as the cache read model.
func (*Client) Bootstrap ¶
func (c *Client) Bootstrap( ctx context.Context, consumer string, stream string, ) (*Snapshot, error)
Bootstrap starts a snapshot-then-stream cycle. It returns the current safe watermark, prior cursor, and a repeatable-read transaction over the public cache tables. Bootstrap DISCARDS every undelivered event at or below the returned SafeSeq when Snapshot.Commit succeeds; the caller must replace its projection in Snapshot.Tx before that commit. Snapshot.Close rolls back and preserves PriorSeq.
Callers must defer Snapshot.Close immediately. Ignoring the lifecycle leaks a pooled connection and retains the cursor row lock, potentially blocking Tail and exhausting the pool (C-S3/C-S4).
func (*Client) Tail ¶
func (c *Client) Tail( ctx context.Context, consumer string, stream string, handler Handler, ) error
Tail continuously pages events with seq > cursor AND seq <= safe_seq, invokes handler inside the cursor transaction, and waits using LISTEN/NOTIFY plus a polling fallback. Migration 0014's after-insert trigger emits ghsync_change_events notifications at commit; correctness never depends on receiving one (C-S2/C-S5/C-P6).
Run exactly one Tail call for each (consumer, stream). A competing tailer returns *ErrCursorContention rather than an opaque PostgreSQL serialization failure. ErrCursorContention and other errors for which IsRetryable is true may be retried with backoff. ErrResyncRequired is terminal for this Tail call and requires Bootstrap plus projection replacement before resuming. Invalid arguments, handler failures, and context cancellation are terminal unless their wrapped cause is independently classified as retryable.
Tail internally retries concurrent cursor first-touch races and all LISTEN failures, including listener connection-pool exhaustion. While LISTEN is unavailable it continues correctness-path polling with bounded reconnect backoff; listener errors are not returned to the caller.
type Config ¶
type Config struct {
// BatchSize is the maximum number of events handled in one cursor
// transaction (C-P6).
BatchSize int
// PollInterval bounds wake latency when LISTEN/NOTIFY is delayed or lost.
PollInterval time.Duration
}
Config controls paging and the correctness-preserving poll fallback.
type ErrCursorContention ¶
type ErrCursorContention struct {
// Consumer is the contended durable consumer name.
Consumer string
// Stream is the contended stream.
Stream string
// contains filtered or unexported fields
}
ErrCursorContention reports that multiple transactions tried to own one durable (consumer, stream) cursor. It is retryable, but the durable fix is to obey Tail's one-tailer rule rather than run competing retry loops.
func (*ErrCursorContention) Error ¶
func (e *ErrCursorContention) Error() string
Error implements error.
func (*ErrCursorContention) Unwrap ¶
func (e *ErrCursorContention) Unwrap() error
Unwrap returns the underlying PostgreSQL serialization error.
type ErrResyncRequired ¶
type ErrResyncRequired struct {
// Consumer is the durable consumer name.
Consumer string
// Stream is the expired stream.
Stream string
// Cursor is the consumer's last committed sequence.
Cursor int64
// PrunedThrough is the greatest sequence known to have been pruned.
PrunedThrough int64
}
ErrResyncRequired is returned when a cursor is behind its stream's pruned horizon. Call Bootstrap and replace the local snapshot before tailing again.
func (*ErrResyncRequired) Error ¶
func (e *ErrResyncRequired) Error() string
Error implements error.
type Event ¶
type Event struct {
// Seq is the global monotonic outbox sequence.
Seq int64
// Stream identifies the event tier, such as entities or work_items.
Stream string
// Kind identifies the additive event variant.
Kind string
// EntityKey is the immutable reference consumers use to fetch current state.
EntityKey string
// OccurredAt is the source transaction's event time.
OccurredAt time.Time
// Payload contains the versioned, additive reference metadata.
Payload json.RawMessage
}
Event is the stable C-S6 change-event envelope. Payload is a versioned reference, never an internal database row image.
type Handler ¶
Handler applies one event using tx. Database effects written through tx commit atomically with the durable cursor advance. Returning an error rolls the entire page back, so a restart receives the page again. C-C6 forbids network I/O in Handler: external effects cannot share this exactly-once transaction and must run after commit with their own idempotency.
type Snapshot ¶
type Snapshot struct {
// SafeSeq is the sequence after which Tail resumes.
SafeSeq int64
// PriorSeq is the durable cursor value Bootstrap replaces on Commit.
PriorSeq int64
// Tx is the snapshot-consistent Postgres transaction over the cache.
Tx pgx.Tx
// contains filtered or unexported fields
}
Snapshot is an open, repeatable-read cache snapshot paired with SafeSeq and PriorSeq. The caller reads public cache tables through Tx and replaces its projection in that same transaction, calls Commit to atomically reset the cursor, and defers Close immediately after Bootstrap.
Bootstrap DISCARDS every undelivered event at or below SafeSeq for this consumer when Commit succeeds. Ignoring both Commit and Close silently skips that cursor reset and leaks a pooled connection while retaining the cursor row lock, which can block the consumer and eventually exhaust the pool. C-C6 forbids network I/O while Tx is open: projection replacement must use only database and CPU work, then commit before any external effect. Tx remains exported for compatibility; prefer Commit and Close for lifecycle management.
func (*Snapshot) Close ¶
Close abandons an uncommitted Snapshot by rolling it back and releasing its pooled connection. Close is idempotent and is safe to defer immediately after Bootstrap; after Commit or a direct Tx finalization it returns nil.
func (*Snapshot) CloseContext ¶
CloseContext abandons an uncommitted Snapshot using a bounded cleanup context derived from ctx while remaining usable after caller cancellation.
Source Files