Documentation
¶
Index ¶
- Variables
- type BarrierOutcome
- type CollisionKind
- type Config
- type Store
- func (s *Store) AcknowledgeTerminals(ctx context.Context, acks []producer.TerminalAck) error
- func (s *Store) CheckpointUnit(ctx context.Context, p UpsertItemParams, unitKey string) error
- func (s *Store) Close() error
- func (s *Store) DestKeyObserved(ctx context.Context, destKey string) (bool, error)
- func (s *Store) ItemDone(ctx context.Context, sourceURI, destURI string) (bool, string, error)
- func (s *Store) LaneStatus(ctx context.Context, lane partition.LaneRef) (producer.LaneStatus, error)
- func (s *Store) MarkDestKeyObserved(ctx context.Context, destKey string) error
- func (s *Store) MarkLaneEOF(ctx context.Context, lanes []partition.LaneRef) error
- func (s *Store) NoteCollision(ctx context.Context, destKey string, kind CollisionKind, ...) error
- func (s *Store) NoteDestKeySource(ctx context.Context, destKey, sourceURI, sourceETag string, sourceSize int64) error
- func (s *Store) OperationCheckpointFingerprint(ctx context.Context, operation string) (string, error)
- func (s *Store) PersistAdmissions(ctx context.Context, admissions []producer.Admission, replay bool) ([]producer.AdmissionRefusal, error)
- func (s *Store) PersistDurableBatch(ctx context.Context, batch producer.DurableBatch) error
- func (s *Store) SetOperationCheckpointIdentity(ctx context.Context, operation, fingerprint string) error
- func (s *Store) SetSourceMetadata(ctx context.Context, provider, bucket, root, sourceURI string) error
- func (s *Store) UnitDone(ctx context.Context, unitKey string) (bool, string, error)
- func (s *Store) UpsertItem(ctx context.Context, p UpsertItemParams) error
- func (s *Store) WriterStats() WriterStats
- type UpsertItemParams
- type WriterStats
Constants ¶
This section is empty.
Variables ¶
var ErrWriterClosed = errors.New("reflow checkpoint writer closed")
ErrWriterClosed is returned to a caller that submits after the coordinator has been gracefully closed. Graceful close is symmetric with the failure path: a late or racing submitter unblocks with this typed error rather than blocking forever on a writer goroutine that has already exited. Well-behaved callers do not submit after Close; this is a fail-fast guard, not a supported ordering.
var ErrWriterFailed = errors.New("reflow checkpoint writer failed")
ErrWriterFailed marks a terminal failure of the checkpoint write coordinator. It is resumable: a run that observes it must abort without treating any pending terminal as durably acknowledged, and a later resume re-drives the affected items. Every mutation waiter blocked on a failing writer unblocks with an error that wraps this sentinel.
Functions ¶
This section is empty.
Types ¶
type BarrierOutcome ¶
type BarrierOutcome uint8
BarrierOutcome classifies how an admitted exec request's post-enqueue barrier completed from the *caller's* observation of awaitExecResult. It is control-flow-neutral telemetry only: it never drives admission, batching, or fail-closed decisions.
Waiter-side, not durable truth: BarrierCanceled (and a cancel-vs-done select race) can count "canceled" even when the write still commits under baseCtx. Writer-side RequestRefusals (from commit results) can therefore diverge from BarrierRefusal under cancel storms. Bind rules and capacity automation must not treat BarrierOK / BarrierCanceled as a commit ledger; prefer WriterStats batch/commit fields and published durability semantics.
const ( // BarrierOK is a published nil result (durably committed success observed // by the waiter). BarrierOK BarrierOutcome = iota // BarrierRefusal is a published non-nil result that is not a terminal // writer lifecycle error (per-request logical refusal under the publish // invariant: success path done values are nil or unwrapped refusal only). BarrierRefusal // BarrierWriterFailed is a terminal writer failure (batch fatal). BarrierWriterFailed // BarrierWriterClosed is a graceful-close lifecycle outcome for a request // that was not published a result. BarrierWriterClosed // BarrierCanceled is a caller-context cancellation while waiting on the // barrier (the write may still commit under baseCtx). BarrierCanceled )
func (BarrierOutcome) String ¶
func (o BarrierOutcome) String() string
String returns a stable label for logs and tests.
type CollisionKind ¶
type CollisionKind string
const ( CollisionDuplicate CollisionKind = "duplicate" CollisionConflict CollisionKind = "conflict" CollisionOverwrite CollisionKind = "overwrite" )
type Config ¶
type Config struct {
Path string
// ElideRawExecSavepoints is an experimental default-off switch: when true,
// the write coordinator omits SAVEPOINT/RELEASE around raw writer.exec
// (SQL statement) mutations. execTx paths always keep savepoints so designed
// request-local refusals still work. Product default remains false — this is
// not a recommended production setting. Optional measure-only override:
// GONIMBUS_REFLOW_ELIDE_RAW_EXEC_SAVEPOINTS=1 (or "true") also enables this
// when Config does not already set it true. Leave unset for normal use.
ElideRawExecSavepoints bool
}
Config opens a reflow checkpoint store.
Writer diagnostics for measure-first work are snapshot-only via Store.WriterStats(); there is no callback observer on Config (C1 deliberately keeps sinks off the sole-writer / post-COMMIT acknowledgement path).
Experimental: the package is Experimental; this surface may change with only an in-release note.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
func (*Store) AcknowledgeTerminals ¶
func (*Store) CheckpointUnit ¶
CheckpointUnit atomically records the downstream terminal and acknowledges its keyed work unit in the same durable transaction.
func (*Store) DestKeyObserved ¶
func (*Store) LaneStatus ¶
func (*Store) MarkDestKeyObserved ¶
func (*Store) MarkLaneEOF ¶
func (*Store) NoteCollision ¶
func (*Store) NoteDestKeySource ¶
func (*Store) OperationCheckpointFingerprint ¶
func (*Store) PersistAdmissions ¶
func (*Store) PersistDurableBatch ¶
func (*Store) SetOperationCheckpointIdentity ¶
func (*Store) SetSourceMetadata ¶
func (*Store) UpsertItem ¶
func (s *Store) UpsertItem(ctx context.Context, p UpsertItemParams) error
func (*Store) WriterStats ¶
func (s *Store) WriterStats() WriterStats
WriterStats returns an immutable aggregate snapshot of checkpoint writer diagnostics (queue/admission, barrier, batch size, batch/tx wall duration, refusals and fatals). It is safe to call concurrently with in-flight work. Queue occupancy is approximate; see WriterStats field docs. A nil Store returns a zero value.
Experimental: the package is Experimental; this surface may change with only an in-release note.
type UpsertItemParams ¶
type WriterStats ¶
type WriterStats struct {
// MaxBatch is the coordinator's admission/coalesce bound (also the exec
// channel capacity).
MaxBatch int
// QueueDepthSamples is how many successful exec admissions contributed a
// depth sample. QueueDepthSum / QueueDepthSamples is the mean sample.
// QueueDepthPeak is the maximum sample observed (approximate occupancy).
QueueDepthSamples int64
QueueDepthSum int64
QueueDepthPeak int64
// Admissions counts successful exec-channel enqueues. AdmissionWaitNanos
// is the sum of admission-select wall times for those enqueues;
// AdmissionWaitMaxNanos is the max single wait. AdmissionBlocked counts
// admits whose pre-send depth sample equaled MaxBatch (channel full /
// backpressured at sample time).
Admissions int64
AdmissionWaitNanos int64
AdmissionWaitMaxNanos int64
AdmissionBlocked int64
// Barrier* cover post-enqueue-to-result waits for exec requests only
// (reads are not counted). Outcome counters partition Barriers.
Barriers int64
BarrierWaitNanos int64
BarrierWaitMaxNanos int64
BarrierOK int64
BarrierRefusal int64
BarrierWriterFailed int64
BarrierWriterClosed int64
BarrierCanceled int64
// Batches is the number of write transactions assembled. BatchSizeSum /
// Batches is mean size in request units (not SQL statements).
// BatchSizeMax is the largest coalesced batch. Coarse histogram buckets:
// 1 | 2–8 | 9–32 | 33–128 | 129+.
Batches int64
BatchSizeSum int64
BatchSizeMax int64
BatchSize1 int64
BatchSize2To8 int64
BatchSize9To32 int64
BatchSize33To128 int64
BatchSize129Plus int64
// Commits counts batch write attempts (success and fatal).
// BatchDuration* is wall time of the full batch write path inside
// c.commit(batch): BeginTx, per-request SQL/savepoints, and tx.Commit().
// It is NOT storage COMMIT alone; bind rules must not treat it as
// SQLite COMMIT sensitivity. CommitFatals counts terminal batch failures.
// RequestRefusals counts per-request logical refusals on successfully
// committed batches (distinct from CommitFatals).
Commits int64
BatchDurationNanos int64
BatchDurationMaxNanos int64
CommitFatals int64
RequestRefusals int64
// SavepointsCreated counts SAVEPOINT statements issued in commit.
// SavepointsElided counts raw-exec requests that skipped SAVEPOINT/RELEASE
// when ElideRawExecSavepoints is enabled (experimental).
SavepointsCreated int64
SavepointsElided int64
}
WriterStats is an immutable aggregate snapshot of checkpoint writer diagnostics. It deliberately excludes paths, SQL text, object keys, URIs, and auth material.
C1 surface is snapshot-only: there is no public callback observer. Always-on atomics update on the existing control path and are read via WriterStats(). That keeps measure-first telemetry off the durability acknowledgement path (post-COMMIT result publish and sole-writer progress are not gated on sinks).
Queue occupancy fields are approximate: depth is sampled as len(execCh) immediately before a successful send. Concurrent drain/admit races mean samples are not a perfect queue integral; peak is a CAS-tracked max of those samples.
Experimental: the package is Experimental; this surface may change with only an in-release note.