storage

package
v0.3.0-20260820034428-... Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 20, 2026 License: Apache-2.0 Imports: 5 Imported by: 0

README

Storage

Pluggable persistence interfaces for SubmitQueue entities (requests, batches, dependents, logs, etc.). Implementations live under extension/storage/<impl>/.

Queue-scoped resolution

Storage follows the extension contract: the queue-scoped store aggregate is resolved per queue through a factory keyed by queue name, mirroring how every decision/action extension resolves its implementation. A resolved aggregate is bound to its queue — entity arguments whose queue disagrees with the binding are rejected, queue-keyed reads are implicitly scoped, and the host wiring decides which backend serves which queue (single shared backend by default).

Three read-model stores are deliberately global rather than queue-scoped, because their lookups start from identifiers that arrive without queue context (a bare request ID or change URI at the status API): the request log, the request summary, and the change-URI mapping. They are injected individually as standalone seams, following the gateway's per-store injection. The queue registry (queueconfig) was never part of this aggregate and stays the registry the factory sits beside.

The classification rule: a store is queue-scoped when every read path authoritatively holds the queue before the first read, and global when any read path begins from an identifier that arrives without queue context. Entity IDs are opaque — no reader may derive the queue from an ID prefix; the queue travels explicitly on payloads and requests.

Optimistic locking contract

Entities that support concurrent mutation carry an int32 Version field. Updates are conditional on the version: the write only succeeds if the persisted version matches the caller's expected version. On mismatch, the implementation returns storage.ErrVersionMismatch, which is declared as a retryable infrastructure error so callers can return it without reclassifying it.

Updates replace every non-primary-key field. Callers must pass a complete authoritative entity loaded from storage or constructed with every persisted field; sparse patch entities can clear unrelated columns. The primary key identifies the row and is not rewritten.

Version arithmetic is owned by the controller, not the store. Versioned update methods take a complete entity plus both oldVersion (the where-clause guard) and newVersion (the value to write):

Update(ctx, request entity.Request, oldVersion, newVersion int32) error

The store writes newVersion rather than the entity's current Version and performs a pure conditional write — it does not compute oldVersion + 1 internally. This keeps the in-memory entity and the persisted row in sync without the storage layer mutating values the caller didn't supply.

Caller pattern
oldVersion := request.Version
newVersion := oldVersion + 1
updated := request
updated.State = newState
if err := store.Update(ctx, updated, oldVersion, newVersion); err != nil {
    return err // request remains unchanged on failure — safe to retry
}
updated.Version = newVersion
request = updated // only after the write succeeded

The candidate-copy pattern keeps the caller-owned entity unchanged if the write fails. Clone slice and map fields before changing their contents so the candidate cannot mutate the original through shared backing storage. The post-success assignment matters whenever the entity is read again later in the same flow. Pre-incrementing in memory before the call is a bug pattern: if the call fails and the caller swallows the error, the in-memory version is now ahead of the database and subsequent updates will fail with ErrVersionMismatch for non-obvious reasons.

Read-after-write consistency

A Get immediately following a successful write (Create/Update) — by the same caller, or a causally-dependent one such as a queue consumer processing a message published after the write committed — must return that write. This is a requirement on every storage implementation, not a condition callers negotiate around: a MySQL primary (including after a promotion) satisfies it, and any other backend (KV, document, etc.) must too.

Controllers must not treat ErrNotFound as "not visible yet, retry." The store interface is intentionally general enough to run over any backend, so a controller has no way to know whether a missing row will appear shortly or does not exist at all — retrying on that assumption just reintroduces, in business logic, the consistency gap the storage contract exists to close. If a Get misses a row that a causally-prior write should already have produced, that is a storage implementation defect: let the error surface as a normal (non-retryable, per platform/errs's default) failure rather than absorbing it with a retryable wrapper.

Key-value contract

Store interfaces are designed for the storage technology space, not for SQL (see the Extensions section of the repo CLAUDE.md): every method must be satisfiable by a plain key-value backend (DynamoDB, Bigtable, an in-memory map) as cheaply as by MySQL. Concretely, a store exposes only get/put/conditional-update by primary key. No lookups by other attributes, no listings filtered server-side, no joins.

The smell test is the index. If implementing a proposed store method in MySQL requires adding a secondary index (KEY idx_*) to the schema, the method is a query-by-attribute in disguise and the contract has left the key-value space — a KV backend would need a global secondary index or a hand-maintained index table to fake it. Treat a new KEY line in a schema diff as a design review flag, not a tuning detail.

Reach for the derived-key pattern instead. When callers need "all X belonging to Y", encode the relationship in the primary key rather than querying for it: derive the key deterministically from the composite identity the caller already holds — for example {parentID}/{hash(child identity)}. Every caller that wants the children can recompute the keys and issue per-key reads; creation under a deterministic key is naturally idempotent (a redelivery finds the existing row); and "at most one row per identity" holds by construction instead of by query discipline.

Domain state is often already the index. Before adding any lookup, check whether an entity the caller already loads enumerates the children — an aggregate that references its parts by ID (e.g. a tree whose paths record their build identities) is the batch→children index, persisted and versioned as domain state. Duplicating that relationship as a database index adds a second source of truth for something the domain already owns.

When neither applies, the reverse lookup is real — give it its own mapping store. In the KV space there is no third mechanism: the only way to look up by an attribute is to make that attribute a primary key somewhere. So promote the relationship to a first-class mapping entity — keyed by the lookup attribute, written by the same flow that creates the source entity with idempotent puts, and rebuildable as a projection if it drifts. ChangeRecord is the in-repo example: it exists so "which requests claimed this change URI" is a by-key read on (queue, URI). QueueBatchState is the same pattern for a mutable attribute: "which batches of this queue are in this state" is a by-key read on (queue, state), maintained as advisory records that move buckets alongside the batch's own state CAS (the shared primitives in submitqueue/core/batch own that protocol) — it replaced BatchStore.GetByQueueAndStates, which was the contract's one query-by-attribute. Unlike a KEY idx_*, the relationship is visible in the contract and portable to any backend.

Decision path

Take the first branch that applies:

  1. Derive — the caller already holds the composite identity → encode it in the primary key. No new state.
  2. Enumerate — an entity already on the caller's path references the children by ID → that aggregate is the index. Escalate to 3 if the list would grow unbounded or take appends from many concurrent writers (a version-contention hotspot under optimistic locking).
  3. Map — a pipeline controller needs the lookup at runtime → a dedicated mapping store keyed by the attribute.

The bar for 3 is a hot-path need: one mapping per access path, never per attribute, and never for ops/debug queries — run those against SQL replicas directly. But don't contort 1–2 to dodge a legitimate 3; a primary key that hashes half the entity's fields, or an aggregate bloated into listing everything, is the same duplication hidden in a worse place.

Documentation

Index

Constants

This section is empty.

Variables

View Source
var ErrAlreadyExists = errors.New("record already exists")

ErrAlreadyExists is returned by storage implementations when attempting to create a record with an ID that already exists.

View Source
var ErrNotFound = errors.New("record not found")

ErrNotFound is returned by storage implementations when the requested record is not found in the database.

View Source
var ErrVersionMismatch = errs.NewRetryableError(errors.New("version mismatch"))

ErrVersionMismatch is returned by storage implementations when the expected entity version does not match the current version of the object. This is used to implement an optimistic locking mechanism, allowing multiple clients to update the same entity concurrently and either retry or implement idempotent operations. It is intrinsically a retryable infrastructure error.

Functions

func IsNotFound

func IsNotFound(err error) bool

IsNotFound returns true if any error in the error chain is a ErrNotFound.

func WrapNotFound

func WrapNotFound(err error) error

WrapNotFound wraps ErrNotFound with the original error from the storage implementation.

Types

type BatchDependentStore

type BatchDependentStore interface {
	// Get retrieves the batch dependent by batch ID.
	// If the batch contains no dependents, the returned BatchDependent will have an empty Dependents list.
	// Returns ErrNotFound if no reverse-index row exists for the batch.
	Get(ctx context.Context, batchID string) (entity.BatchDependent, error)

	// Create creates a new batch dependent.
	// Returns ErrAlreadyExists if the entry already exists for the given batch ID.
	Create(ctx context.Context, batchDependent entity.BatchDependent) error

	// Update replaces the non-key fields of a batch dependent and persists newVersion
	// if the current persisted version matches oldVersion. If versions do not match, returns ErrVersionMismatch.
	// Version arithmetic is owned by the caller; the store performs a pure conditional write.
	Update(ctx context.Context, batchDependent entity.BatchDependent, oldVersion, newVersion int32) error
}

BatchDependentStore is an interface that defines methods for managing batch dependent information in the database.

A BatchDependent is a reverse index ("batches that depend on me") paired one-to-one with a Batch. The batch-creation flow creates this row while the Batch is Creating and before making the Batch eligible for pipeline processing. A Creating Batch can briefly exist without its row; every Batch that reaches Created is guaranteed to have one.

type BatchStore

type BatchStore interface {
	// Get retrieves a batch by ID. Returns ErrNotFound if the batch is not found.
	Get(ctx context.Context, id string) (entity.Batch, error)

	// Create creates a new batch. The batch must have a unique ID already assigned.
	// Returns ErrAlreadyExists if a batch with the same ID already exists.
	Create(ctx context.Context, batch entity.Batch) error

	// Update replaces every non-key field of a batch and writes newVersion
	// if the current persisted version matches oldVersion. If versions do not match, returns ErrVersionMismatch.
	// Version arithmetic is owned by the caller; the store performs a pure conditional write.
	Update(ctx context.Context, batch entity.Batch, oldVersion, newVersion int32) error
}

BatchStore is an interface that defines methods for managing batches in the database.

type BuildStore

type BuildStore interface {
	// Get retrieves a build by ID. Returns ErrNotFound if the build is not found.
	Get(ctx context.Context, id string) (entity.Build, error)

	// Create creates a new build. The build must have a unique ID already assigned.
	// Returns ErrAlreadyExists if a build with the same ID already exists.
	Create(ctx context.Context, build entity.Build) error

	// Update replaces all non-key fields of a build.
	Update(ctx context.Context, build entity.Build) error
}

BuildStore is an interface that defines methods for managing builds in the database.

type ChangeStore

type ChangeStore interface {
	// Create persists a single ChangeRecord (identity + Details) in one write. A
	// primary-key conflict on (Queue, URI, RequestID) is silently ignored, which makes
	// the call idempotent under queue redeliveries of the same request (first write
	// wins). Records belonging to different requests do not conflict on the PK —
	// cross-request overlap is detected by GetByURI, not by Create.
	Create(ctx context.Context, record entity.ChangeRecord) error

	// GetByURI returns every ChangeRecord the bound queue holds for the given
	// URI. Multiple requests can have claimed the same URI over time, so the
	// slice may have any number of entries; an empty slice means no claim has
	// ever been recorded for this URI in this queue.
	//
	// The store does not filter by request_id or by the owning request's
	// state — callers that want to skip self filter by RequestID, and callers
	// that want only live owners consult RequestStore for liveness.
	GetByURI(ctx context.Context, uri string) ([]entity.ChangeRecord, error)
}

ChangeStore manages per-URI claim records for in-flight land requests. Each row records that a specific URI was claimed by a specific request, scoped to a queue. The (Queue, URI, RequestID) triple is the immutable identity of a record, and a record's Details are captured at claim time and never updated — records are immutable.

The interface is intentionally per-record / per-URI so that any backend (SQL, DynamoDB, Bigtable, …) can implement it without needing batch-atomicity or multi-key query support. Callers loop when they have multiple URIs to claim or check; the typical request has a small number of URIs (a single PR or a short stack), so the loop overhead is negligible.

type Config

type Config struct {
	// QueueName is the name of the queue whose data the resolved Storage is
	// scoped to.
	QueueName string
}

Config identifies the queue a Storage instance is resolved for. Like every other extension config, it carries only the queue name — everything an implementation needs beyond that is injected at construction by the integrator.

type Factory

type Factory interface {
	// For returns the Storage aggregate bound to the queue named in config.
	For(config Config) (Storage, error)
}

Factory resolves the queue-scoped Storage aggregate for a queue. Mirrors the extension contract: the host wiring decides which backend serves which queue; implementations bind the queue over their backend so a resolved instance can only read and write that queue's data.

type PathBuildStore

type PathBuildStore interface {
	// Get resolves an attempt to its build.
	// Returns ErrNotFound if no build has been recorded for the attempt.
	Get(ctx context.Context, pathID string, attempt int) (entity.PathBuild, error)

	// Create records the build for an attempt, permanently.
	// Returns ErrAlreadyExists if the attempt already has a build, which means
	// a concurrent dispatch recorded its own first.
	Create(ctx context.Context, pathBuild entity.PathBuild) error
}

PathBuildStore resolves one attempt of one speculation path to the build started for it, keyed by (path ID, attempt).

It exists because the runner chooses the build ID: a caller holding a path and an attempt cannot derive it, and there is no lookup by attribute in this contract. Promoting the relationship to a keyed record is the mechanism for a reverse lookup here, not a workaround.

A record is write-once and complete from the start: it is created only once the runner has named the build, and never updated afterwards. It answers two questions — absent means no build is recorded for the attempt, present names the attempt's build permanently. A retried path is a new attempt under a different key.

Because creation is the only write, a duplicate Create is how concurrent dispatches for the same attempt are decided: the first insert wins, and ErrAlreadyExists tells the loser the attempt's build is someone else's.

type QueueBatchStateStore

type QueueBatchStateStore interface {
	// List returns every record filed under the bound queue's given state bucket.
	// An empty slice means the bucket is empty. Order is unspecified.
	List(ctx context.Context, state entity.BatchState) ([]entity.QueueBatchState, error)

	// Put persists a record. The record's Queue must match the instance's bound
	// queue. Writing an already-existing (queue, state, batchID) record is a no-op
	// success, so the call is idempotent under redeliveries.
	Put(ctx context.Context, record entity.QueueBatchState) error

	// Delete removes the bound queue's record identified by (state, batchID).
	// Deleting an absent record is a no-op success, so the call is idempotent
	// under redeliveries.
	Delete(ctx context.Context, state entity.BatchState, batchID string) error
}

QueueBatchStateStore manages the per-queue membership records that file each in-queue batch under a lifecycle state bucket, so "the batches of queue Q filed under state S" is a single read keyed by (queue, state) — a primary-key prefix on any backend, with no secondary index or server-side filtering required.

Records are advisory: the authoritative state is on the Batch entity, and a record may transiently file a batch under a bucket it has already left. Readers therefore treat a listing as a set of candidate batch IDs — they load each Batch by key and classify it by its own State, never by the bucket the record was found in.

The interface is intentionally per-state and per-record so that any backend (SQL, DynamoDB, Bigtable, …) can implement it without multi-key queries or batch atomicity. Callers loop over the states they care about; a batch moves buckets via Put of the new record followed by Delete of the old one, which keeps at least one record visible throughout. All writes are idempotent so queue redeliveries can safely repeat them.

type RequestBatchStore

type RequestBatchStore interface {
	// GetByRequestID retrieves every batch association for a request.
	GetByRequestID(ctx context.Context, requestID string) ([]entity.RequestBatch, error)

	// Create inserts an immutable association. Returns ErrAlreadyExists if the request and batch are already associated.
	Create(ctx context.Context, association entity.RequestBatch) error
}

RequestBatchStore persists immutable associations between requests and the batch attempts containing them.

type RequestLogStore

type RequestLogStore interface {
	// Insert appends a new request log record. Timestamps should be generated by the caller and not modified by the implementation.
	Insert(ctx context.Context, log entity.RequestLog) error

	// List retrieves all request log records for a given request ID, ordered by timestamp ascending.
	// Returns ErrNotFound if no records exist for the given request ID.
	List(ctx context.Context, requestID string) ([]entity.RequestLog, error)
}

RequestLogStore is an interface that defines methods for managing request log records in an append-only database. Request logs are used to reconcile request statuses with eventual consistency into a separate database from RequestStore.

type RequestQueueSummaryCursor

type RequestQueueSummaryCursor struct {
	// ReceivedAtMs is the receipt timestamp of the last item from the previous page.
	ReceivedAtMs int64
	// RequestID is the request ID of the last item from the previous page.
	RequestID string
}

RequestQueueSummaryCursor is the exclusive keyset boundary for a descending queue-summary query.

type RequestQueueSummaryQuery

type RequestQueueSummaryQuery struct {
	// ReceivedAtOrAfterMs is the inclusive lower receipt-time bound.
	ReceivedAtOrAfterMs int64
	// ReceivedBeforeMs is the exclusive upper receipt-time bound.
	ReceivedBeforeMs int64
	// Cursor is an exclusive continuation boundary when HasCursor is true.
	Cursor RequestQueueSummaryCursor
	// HasCursor selects whether Cursor participates in the query.
	HasCursor bool
	// Limit is the maximum number of rows returned and must be positive.
	Limit int
}

RequestQueueSummaryQuery specifies one bounded queue-summary page query against the bound queue's partition.

type RequestQueueSummaryStore

type RequestQueueSummaryStore interface {
	// Create inserts summary and returns ErrAlreadyExists when its full primary key already exists.
	// The summary's Queue must match the instance's bound queue.
	Create(ctx context.Context, summary entity.RequestQueueSummary) error

	// Get returns the bound queue's row identified by (receivedAtMs, requestID), or ErrNotFound when absent.
	Get(ctx context.Context, receivedAtMs int64, requestID string) (entity.RequestQueueSummary, error)

	// Update conditionally replaces all non-key fields when the persisted projection version equals oldVersion.
	// The store writes newVersion exactly as supplied and returns ErrVersionMismatch when the guard does not match.
	Update(ctx context.Context, summary entity.RequestQueueSummary, oldVersion, newVersion int32) error

	// List returns at most query.Limit rows ordered by received_at_ms descending, then request_id descending.
	List(ctx context.Context, query RequestQueueSummaryQuery) ([]entity.RequestQueueSummary, error)
}

RequestQueueSummaryStore persists the queue-ordered request projection.

type RequestStore

type RequestStore interface {
	// Get retrieves a land request by ID. Returns ErrNotFound if the request is not found.
	Get(ctx context.Context, id string) (entity.Request, error)

	// Create creates a new land request. The request must have a unique ID already assigned.
	// Returns ErrAlreadyExists if a request with the same ID already exists.
	Create(ctx context.Context, request entity.Request) error

	// Update replaces every non-key field of a land request and writes newVersion if the current persisted version matches oldVersion.
	// If versions do not match, returns ErrVersionMismatch. Version arithmetic is owned by the caller; the store performs a pure conditional write.
	Update(ctx context.Context, request entity.Request, oldVersion, newVersion int32) error
}

RequestStore is an interface that defines methods for managing land requests in the database.

type RequestSummaryStore

type RequestSummaryStore interface {
	// Create inserts summary and returns ErrAlreadyExists when RequestID already exists.
	// The caller owns retry identity and decides whether an existing row is an identical retry or a conflict.
	Create(ctx context.Context, summary entity.RequestSummary) error

	// Get returns the summary for requestID, or ErrNotFound when absent.
	Get(ctx context.Context, requestID string) (entity.RequestSummary, error)

	// Update conditionally replaces every non-key field when the persisted projection version equals oldVersion.
	// The store writes newVersion exactly as supplied and returns ErrVersionMismatch when the guard does not match.
	Update(ctx context.Context, summary entity.RequestSummary, oldVersion, newVersion int32) error
}

RequestSummaryStore persists the authoritative request-ID materialized view.

type RequestURIStore

type RequestURIStore interface {
	// Create inserts mapping and returns ErrAlreadyExists when its full primary key already exists.
	Create(ctx context.Context, mapping entity.RequestURI) error

	// ListByURI returns at most limit mappings ordered by received_at_ms descending, then request_id descending.
	ListByURI(ctx context.Context, changeURI string, limit int) ([]entity.RequestURI, error)
}

RequestURIStore persists the immutable change-URI reverse mapping.

type SpeculationPathSetStore

type SpeculationPathSetStore interface {
	// Get retrieves a head's path set, where head is the head batch's ID.
	// Returns ErrNotFound if the head has no set yet, which is the normal state
	// for a batch nothing has speculated on.
	Get(ctx context.Context, head string) (entity.SpeculationPathSet, error)

	// Create stores a head's first path set.
	// Returns ErrAlreadyExists if the head already has one.
	Create(ctx context.Context, set entity.SpeculationPathSet) error

	// Update replaces the stored set with set and writes newVersion, but only if
	// the persisted version still matches oldVersion. If it does not, returns
	// ErrVersionMismatch and writes nothing.
	//
	// The whole entity goes in rather than the fields being changed: a set is
	// replaced wholesale, so this is a conditional put on a key — the primitive
	// every backend offers directly, instead of a field-level update each
	// non-SQL backend would have to emulate with a read-modify-write.
	//
	// set.Version is ignored. oldVersion is the guard and newVersion is the
	// value written, so version arithmetic stays with the caller: compute
	// newVersion, call, and assign it to the in-memory set only once this
	// returns nil.
	Update(ctx context.Context, set entity.SpeculationPathSet, oldVersion, newVersion int32) error
}

SpeculationPathSetStore persists one head batch's chosen speculation paths.

A set is keyed by its head batch ID within the bound queue and versioned as a whole: every path in it shares that head, and the set is the unit of both replacement and optimistic locking. There is no lookup by anything but the head — callers that need a queue's live sets enumerate the heads from the batch listing they already hold and read each set by key.

type Storage

type Storage interface {
	// GetRequestStore returns the RequestStore instance.
	GetRequestStore() RequestStore

	// GetRequestBatchStore returns the RequestBatchStore instance.
	GetRequestBatchStore() RequestBatchStore

	// GetChangeStore returns the ChangeStore instance.
	GetChangeStore() ChangeStore

	// GetBatchStore returns the BatchStore instance.
	GetBatchStore() BatchStore

	// GetBatchDependentStore returns the BatchDependentStore instance.
	GetBatchDependentStore() BatchDependentStore

	// GetQueueBatchStateStore returns the QueueBatchStateStore instance.
	GetQueueBatchStateStore() QueueBatchStateStore

	// GetBuildStore returns the BuildStore instance.
	GetBuildStore() BuildStore

	// GetSpeculationPathSetStore returns the SpeculationPathSetStore instance.
	GetSpeculationPathSetStore() SpeculationPathSetStore

	// GetPathBuildStore returns the PathBuildStore instance.
	GetPathBuildStore() PathBuildStore

	// GetRequestQueueSummaryStore returns the RequestQueueSummaryStore instance.
	GetRequestQueueSummaryStore() RequestQueueSummaryStore

	// GetRequestSummaryStore returns the RequestSummaryStore instance.
	GetRequestSummaryStore() RequestSummaryStore

	// GetRequestLogStore returns the RequestLogStore instance.
	GetRequestLogStore() RequestLogStore

	// GetRequestURIStore returns the RequestURIStore instance.
	GetRequestURIStore() RequestURIStore
}

Storage aggregates the queue-scoped entity stores into a single injectable dependency. An instance is resolved per queue through Factory and is bound to that queue: entity arguments whose Queue field disagrees with the binding are rejected, and reads never surface another queue's records.

Directories

Path Synopsis
Package mock is a generated GoMock package.
Package mock is a generated GoMock package.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL