Documentation
¶
Index ¶
- type DecisionRow
- type OutboxRow
- type RemoteSyncCommitRecord
- type Store
- func (s *Store) AckOutbox(id, owner string) error
- func (s *Store) AppendDecision(d contract.Decision) error
- func (s *Store) AppendEvent(ev contract.Event) (int64, error)
- func (s *Store) ClaimOutbox(owner string, ttl time.Duration, kinds ...string) ([]OutboxRow, error)
- func (s *Store) Close() error
- func (s *Store) CursorsByPrefix(prefix string) (map[string]int64, error)
- func (s *Store) DecisionCount() int
- func (s *Store) DecisionsAfter(rowid int64) ([]DecisionRow, error)
- func (s *Store) DecisionsForActor(actor contract.ActorID) ([]contract.Decision, error)
- func (s *Store) DeferralCount(correlationID string) int
- func (s *Store) DeleteAckedOutbox(kind string) (int, error)
- func (s *Store) DispatchTx(events []contract.Event, cursorName string, seq int64) error
- func (s *Store) GetCursor(name string) int64
- func (s *Store) GetResource(ref contract.ResourceRef) (contract.Version, map[string]any, error)
- func (s *Store) GetVersion(ref contract.ResourceRef) (contract.Version, error)
- func (s *Store) IngestObservation(env contract.ObservationEnvelope) (int64, bool, error)
- func (s *Store) MarkSyncCommitStatus(originReplicaID, localDecisionID string, ref contract.ResourceRef, ...) error
- func (s *Store) MaxDecidedSeq() int64
- func (s *Store) PendingEvents(afterSeq int64) ([]contract.Event, error)
- func (s *Store) PendingSyncCommits() ([]contract.LocalCommit, error)
- func (s *Store) RecordRemoteSyncCommit(remotePeerID string, commit contract.LocalCommit, receivedAt string) (RemoteSyncCommitRecord, error)
- func (s *Store) RemoteSyncCommitCount() (int64, error)
- func (s *Store) RemoteSyncCommitsAfter(afterSeq int64, excludeOriginReplicaID string, scopes []contract.ResourceRef, ...) ([]RemoteSyncCommitRecord, int64, error)
- func (s *Store) ReplicaID() (string, error)
- func (s *Store) SetCursor(name string, seq int64) error
- func (s *Store) SyncCommitCounts() (SyncCommitCounts, error)
- func (s *Store) WithTx(fn func(*Tx) error) error
- type SyncCommitCounts
- type Tx
- func (t *Tx) AppendDecisionTx(d contract.Decision) error
- func (t *Tx) AppendEvent(ev contract.Event) error
- func (t *Tx) AppendEventReturningSeq(ev contract.Event) (int64, error)
- func (t *Tx) CASUpdate(ref contract.ResourceRef, basedOn contract.Version, fields map[string]any) (bool, error)
- func (t *Tx) CreateResource(ref contract.ResourceRef, fields map[string]any) error
- func (t *Tx) EnqueueOutbox(row OutboxRow) error
- func (t *Tx) ReadResource(ref contract.ResourceRef) (contract.Version, map[string]any, error)
- func (t *Tx) ReadVersion(ref contract.ResourceRef) (contract.Version, error)
- func (t *Tx) RecordSyncCommitsTx(d contract.Decision, syncable map[contract.ResourceKind]bool) error
- func (t *Tx) SetCursor(name string, seq int64) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DecisionRow ¶
DecisionRow is a decision plus its durable append order (the implicit rowid). The server's side-effect sink advances by Rowid, so it can RE-DERIVE invalidations/diagnostics from the decision log after a crash.
type OutboxRow ¶
type OutboxRow struct {
ID string
Kind string
EventSeq int64
Target string
Payload string
Status string
IdempotencyKey string
Attempts int
LeaseOwner string
LeaseUntil int64 // unix seconds; 0 = unleased
}
OutboxRow is one pending external effect (a projection invalidation, a job to run). The outbox is the transactional-outbox substrate (S2: an invalidation is produced from the durable decision log under a sink cursor, so it survives a crash between the decision commit and its side-effect — RECOVERABLE, not lost; S4: delivery is at-least-once with a per-row lease + an idempotency key, NEVER exactly-once).
type RemoteSyncCommitRecord ¶
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
func (*Store) AckOutbox ¶
AckOutbox marks a delivered effect done. It must be acked by the lease holder (a stale worker whose lease expired and was reclaimed cannot ack the new holder's row) — an ack of a row not held by owner is an error.
func (*Store) AppendDecision ¶
AppendDecision writes a decision in its own txn (used for non-accepted ops — nothing to be atomic with).
func (*Store) ClaimOutbox ¶
ClaimOutbox leases every currently-claimable row (not acked, and either unleased or with an expired lease) to owner for ttl, bumping attempts, and returns them. The single writer connection serializes claims, so two workers never both win the same row (S4 delivery lease). Rows are read fully before the UPDATE so the single connection is not held by an open cursor during the writes. If kinds is non-empty, ONLY rows of those kinds are claimed — a delivery worker must lease only the rows it actually delivers (one kind's worker never grabs another kind's rows); empty kinds claims every kind.
func (*Store) CursorsByPrefix ¶
CursorsByPrefix reads every durable cursor whose name starts with prefix, keyed by the name REMAINDER (the per-replica suffix). The hub's status surface uses it for the last-served-cursor-per-replica counter; it is read-only bookkeeping, never an ordering source.
func (*Store) DecisionCount ¶
func (*Store) DecisionsAfter ¶
func (s *Store) DecisionsAfter(rowid int64) ([]DecisionRow, error)
DecisionsAfter returns decisions appended after rowid, in append order. It lets the server produce a decision's side-effects (S2 invalidation / S7 diagnostic) idempotently from the durable log rather than only from a single RunOnce return — closing the crash window between the decision commit and its effects.
func (*Store) DecisionsForActor ¶
DecisionsForActor returns this actor's deferred decisions (the pull-feedback source, Invariant #8).
func (*Store) DeferralCount ¶
DeferralCount returns how many REBASE deferrals a CorrelationID has accumulated in the durable log. It is the liveness-escalation counter (Invariant #10) derived from the decision log rather than held in memory, so escalation survives a process restart exactly as the cursor does. It counts ONLY next_action='rebase' deferrals, so an unrelated human_review deferral (from a defer_to_human / auto_merge_disjoint pass that shares the CorrelationID) does NOT pre-seed the count and trigger premature escalation.
Scope note: for the reconciler path — the SOLE production caller of Kernel.Apply — this reproduces the removed in-memory map's predicate exactly (rebase-deferrals are only produced inside RunOnce). A direct (non-reconciler) Apply sharing a CorrelationID would also contribute to this durable count; that is not reachable today and is the intended behaviour for any future direct path (e.g. CLI-inline, Invariant #17).
func (*Store) DeleteAckedOutbox ¶
DeleteAckedOutbox prunes terminally-acked rows of one kind. Acked rows are never re-read (ClaimOutbox excludes them), so a consumer that acks without pruning grows the outbox by one dead row per accepted decision for the life of the project. Returns how many rows were pruned.
func (*Store) DispatchTx ¶
DispatchTx atomically appends all proposed events produced from ONE observed event AND advances the dispatch cursor past it. All-or-nothing (Invariant R6 / finding #1): a crash can never leave appended proposals with an un-advanced cursor (which would re-dispatch and duplicate).
func (*Store) GetCursor ¶
GetCursor returns a named durable cursor (0 if unset). The runtime's dispatch position lives here, the way the reconciler's decision position is derived from MaxDecidedSeq — both make the loop restart-safe.
func (*Store) GetResource ¶
GetResource returns a resource's current version AND decoded field content (review #5). The content digest (D8/S10), budget reserve (S6), and lease TTL read all need the fields, not just the version. Absent -> (0, nil, nil), consistent with GetVersion.
func (*Store) GetVersion ¶
func (*Store) IngestObservation ¶
IngestObservation appends an observation exactly once per (Source,ExternalID): a retried envelope returns the original seq with dup=true and never double-appends (S1). The dedupe lookup, the event append (with its LSN via AppendEventReturningSeq, review #6), and the dedupe-row insert are ONE transaction; the single writer connection serializes concurrent ingests, so the SELECT-then-INSERT cannot race. The server stamps env.Event.Actor from the authenticated principal BEFORE calling (D7) — the store trusts the envelope.
func (*Store) MarkSyncCommitStatus ¶
func (*Store) MaxDecidedSeq ¶
MaxDecidedSeq returns the highest event ingest_seq that already has a decision (0 if none, or if only direct non-event ops have been applied — those carry ingest_seq 0). The decision log IS the reconciler's durable cursor: a decision row means that event was consumed, so a fresh Reconciler resumes from here instead of re-reading from 0 (Invariant #9 — exactly-once over a contiguous prefix).
func (*Store) PendingEvents ¶
func (*Store) PendingSyncCommits ¶
func (s *Store) PendingSyncCommits() ([]contract.LocalCommit, error)
func (*Store) RecordRemoteSyncCommit ¶
func (s *Store) RecordRemoteSyncCommit(remotePeerID string, commit contract.LocalCommit, receivedAt string) (RemoteSyncCommitRecord, error)
func (*Store) RemoteSyncCommitCount ¶
RemoteSyncCommitCount is the hub's "commits received" counter: the total commits accepted into the append-only sync_remote_commits log (rejected/conflicted commits are never inserted).
func (*Store) RemoteSyncCommitsAfter ¶
func (s *Store) RemoteSyncCommitsAfter(afterSeq int64, excludeOriginReplicaID string, scopes []contract.ResourceRef, limit int) ([]RemoteSyncCommitRecord, int64, error)
func (*Store) SyncCommitCounts ¶
func (s *Store) SyncCommitCounts() (SyncCommitCounts, error)
type SyncCommitCounts ¶
type Tx ¶
type Tx struct {
// contains filtered or unexported fields
}
func (*Tx) AppendDecisionTx ¶
AppendDecisionTx writes a decision INSIDE a caller's txn (used for accepted ops — crash-safe atomicity, Invariant #7).
func (*Tx) AppendEvent ¶
Tx-scoped variants for the atomic dispatch transaction.
func (*Tx) AppendEventReturningSeq ¶
AppendEventReturningSeq appends an event INSIDE a caller's txn and returns its durable LSN (events.rowid). IngestObservation (S1) needs append + LSN-read in one transaction so the dedupe row records the same seq.
func (*Tx) CreateResource ¶
func (*Tx) EnqueueOutbox ¶
EnqueueOutbox inserts a pending effect INSIDE a caller's txn so it commits atomically with the decision (S2). A duplicate idempotency key is a silent no-op (S4 — at-least-once delivery must never enqueue the same effect twice). An empty key is stored as NULL: multiple NULLs are distinct in a UNIQUE index, so keyless rows (e.g. invalidations) never collide with each other.
func (*Tx) ReadResource ¶
ReadResource is GetResource inside a caller's txn — required by the read-modify-write lease/budget claims (S5/S6), which read the current version+fields and CAS in the same transaction.
Source Files