Documentation
¶
Overview ¶
Package engine turns customer-database change events into recomputed Database Sync docs. It owns the dirty-key set, the per-key coalescing that keeps one hot row from stampeding the customer's database, and the mapping between doc channels and query definitions. The load-shedding discipline is ported from the internal gateway's dispatcher (services/realtime/internal/ dispatcher): one refetch per dirty key per round, in-flight dedup so a key is never computed twice concurrently, and a re-dirty during compute just schedules one more round.
Index ¶
- func BuildChannel(queryName string, values []string) string
- type Engine
- func (engine *Engine) DirtyDepth() int64
- func (engine *Engine) HandleEvent(event Event)
- func (engine *Engine) Start(ctx context.Context)
- func (engine *Engine) SyncLive(channels []string, complete bool)
- func (engine *Engine) Touch(channel string)
- func (engine *Engine) Warm(channel string)
- func (engine *Engine) WatchedTables() []string
- type Event
- type Executor
- func (executor *Executor) Close()
- func (executor *Executor) Ping(ctx context.Context) error
- func (executor *Executor) Pool() *pgxpool.Pool
- func (executor *Executor) RunDoc(ctx context.Context, sql string, args []any) (json.RawMessage, error)
- func (executor *Executor) RunKeys(ctx context.Context, sql string, args []any, maxKeys int) ([][]any, error)
- type Publisher
- type Runner
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func BuildChannel ¶
BuildChannel returns the db: doc channel for a query and its param values.
Types ¶
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine coalesces change events into doc recomputes. One Engine per source.
func (*Engine) DirtyDepth ¶
DirtyDepth reports the pending work backlog (docs awaiting a recompute plus queued reverse-index lookups), for heartbeats.
func (*Engine) HandleEvent ¶
HandleEvent maps one row change to affected docs and marks them dirty. It never touches the customer database: keysSql lookups are queued for the workers, so per-event cost is a map lookup plus set inserts and the replication stream is never stalled behind SQL.
func (*Engine) SyncLive ¶
SyncLive reconciles the live set against a liveness poll: the fleet's list of db: channels with subscribers. Listed channels are (re)leased, unlisted ones are dropped unless they were warmed or touched within touchGrace (the poll snapshot may predate a fresh subscriber). An incomplete poll (the fleet view was truncated or unavailable) only adds, never drops: when we cannot see the whole truth, keeping a dead doc warm beats going silent on a watched one.
A listed channel that was not already live is also marked dirty, mirroring Touch: it may have skipped changes while it was cold (or wrongly dropped by an earlier poll). The startup poll takes this path for every doc because the live set starts empty, giving each one recompute per process start; that heals any change the reader acked but the agent never applied before a crash, and lets docs converge after a WAL safety-valve detach + restart.
func (*Engine) Touch ¶
Touch handles a watcher signal: someone started subscribing to a doc that is still retained. A live doc just gets its lease renewed (it has been receiving every change, so it is current). A doc that was NOT live may have skipped changes while cold, so it is recomputed too.
func (*Engine) Warm ¶
Warm handles a warm request: a client attached to a channel whose doc is missing from the retained store. The channel becomes live and is computed as soon as a worker is free.
func (*Engine) WatchedTables ¶
WatchedTables returns the bare (public-schema) table names the definitions watch, in the shape the replication reader's publication reconcile expects.
type Event ¶
type Event struct {
// Table is the fully qualified "schema.table" name.
Table string
OldData map[string]any
NewData map[string]any
}
Event is one row change from the replication stream, reduced to what the engine needs.
type Executor ¶
type Executor struct {
// contains filtered or unexported fields
}
Executor runs customer SQL on the customer's own database. Every session is pinned read-only with a statement timeout at connect time, so no customer query, hostile or accidental, can write or run forever. This session pinning, not SQL inspection, is the real safety layer.
func NewExecutor ¶
func NewExecutor(ctx context.Context, dsn string, statementTimeout time.Duration, maxDocBytes int) (*Executor, error)
NewExecutor connects the customer-DB query pool. dsn is the plain (non replication) connection string.
func (*Executor) Pool ¶
Pool exposes the underlying pool for agent housekeeping queries (slot lag, slot drop). Doc and keys queries go through RunDoc/RunKeys.
func (*Executor) RunDoc ¶
func (executor *Executor) RunDoc(ctx context.Context, sql string, args []any) (json.RawMessage, error)
RunDoc executes a doc query: exactly one column, at most one row. Zero rows return JSON null (the doc does not exist). The result must marshal to JSON within the doc size cap.
func (*Executor) RunKeys ¶
func (executor *Executor) RunKeys(ctx context.Context, sql string, args []any, maxKeys int) ([][]any, error)
RunKeys executes a keysSql reverse index and returns up to maxKeys rows, each row's values in column order (column i feeds query param $i+1). Hitting the cap is an error by design: a truncated key set would silently leave some docs stale, so the customer must tighten the query or raise maxKeys instead.
type Publisher ¶
type Publisher interface {
PublishDoc(ctx context.Context, channel string, doc json.RawMessage) error
}
Publisher delivers a computed doc to its db: channel on foony.
type Runner ¶
type Runner interface {
RunDoc(ctx context.Context, sql string, args []any) (json.RawMessage, error)
// RunKeys returns each result row's values in column order (column i is
// the value for param $i+1).
RunKeys(ctx context.Context, sql string, args []any, maxKeys int) ([][]any, error)
}
Runner executes customer SQL: the doc query and keysSql reverse indexes.
Source Files