Documentation
¶
Index ¶
- func PublishBatchEvents(ctx context.Context, registry consumer.TopicRegistry, queue string, ...) error
- func PublishBatchLogs(ctx context.Context, registry consumer.TopicRegistry, queue string, ...) error
- func PublishLog(ctx context.Context, registry consumer.TopicRegistry, ...) error
- type CurrentState
- type Materializer
- type TerminationOutcome
- type TerminationResult
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func PublishBatchEvents ¶
func PublishBatchEvents(ctx context.Context, registry consumer.TopicRegistry, queue string, requestIDs []string, event entity.RequestEvent, occurrence string, metadata map[string]string) error
PublishBatchEvents publishes an event log entry for each request ID in the batch to the log topic. It is the event counterpart of PublishBatchLogs and shares its partitioning and occurrence semantics.
Separate rather than a flag on PublishBatchLogs because the two carry different vocabularies: a caller reporting build progress cannot reach for a status, and a caller reporting a status cannot accidentally publish something the summary will refuse to project.
func PublishBatchLogs ¶
func PublishBatchLogs(ctx context.Context, registry consumer.TopicRegistry, queue string, requestIDs []string, status entity.RequestStatus, occurrence string, metadata map[string]string) error
PublishBatchLogs publishes a status log entry for each request ID in the batch to the log topic. Each entry uses the request ID as the partition key to ensure per-request ordering. queue scopes every entry: a request ID is only unique within its own queue. occurrence names this occurrence of the status and is shared by every entry in the fan-out, so the whole batch dedupes together on a redelivery; see PublishLog.
Entries carry no request version: a status published for a whole batch reports something that happened to the batch, not a transition of any one request's state machine.
func PublishLog ¶
func PublishLog(ctx context.Context, registry consumer.TopicRegistry, logEntry entity.RequestLog, partitionKey string, occurrence string) error
PublishLog publishes a single request log entry to the log topic for async persistence. The partitionKey ensures ordering of log entries for the same request; typically set to the request ID.
The message ID is scoped to (requestID, status, occurrence) so that the queue's (topic, partition_key, id) unique index dedupes retries of the same logical log event (same delivery re-processed) without rejecting distinct statuses for the same request (e.g. "started" emitted by the start controller and "cancelled" emitted later by the cancel controller).
occurrence is a stable discriminator naming this occurrence of the status. Empty means the status happens at most once per request, and the id collapses to (requestID, status) — every repeat is then dropped at publish time, which is what a terminal status wants. A status that legitimately recurs (a request builds again each time speculation re-plans) must pass something that is stable across redeliveries of one occurrence but differs between occurrences: the build ID, or the batch and path the entry is about. Deriving it from the wall clock or a random value would defeat the dedupe entirely.
Types ¶
type CurrentState ¶
type CurrentState struct {
// Status is the current request status obtained from the request log.
Status entity.RequestStatus
// LastError is the last error associated with the current status.
LastError string
// Metadata is the metadata associated with the current status.
Metadata map[string]string
}
CurrentState holds the current request status obtained from the request log. It is eventually consistent with the request status in the request store. It might take some time to converge, typically no more than a few seconds.
func GetCurrentStateFromRequestLog ¶
func GetCurrentStateFromRequestLog(ctx context.Context, store storage.RequestLogStore, requestID string) (CurrentState, error)
GetCurrentStateFromRequestLog returns the current reconciled state for a request by reading the request log. Returns ErrNotFound if the request ID has no records in the log database. The state is eventually consistent with the request status in the request store. It might take some time to converge, typically no more than a few seconds.
type Materializer ¶
type Materializer struct {
// contains filtered or unexported fields
}
Materializer appends request logs and projects the winning public request state. It owns winner selection, optimistic concurrency, and public projection repair. Every store it touches is queue-scoped, so each call resolves the aggregate once from the queue carried on the log being persisted.
func NewMaterializer ¶
func NewMaterializer(stores storage.Factory) *Materializer
NewMaterializer creates a request read-model materializer.
func (*Materializer) PersistLog ¶
func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) error
PersistLog appends one audit log and materializes its winning state. Projection errors are returned so queue deliveries are retried rather than silently dropping the side write. Because the append happens first, retrying after a projection failure may retain another copy of the event in History.
type TerminationOutcome ¶
type TerminationOutcome int
TerminationOutcome describes what TerminateRequest did to the request. The zero value (TerminationOutcomeUnknown) is only produced alongside a non-nil error.
const ( // TerminationOutcomeUnknown is the zero value, produced only when TerminateRequest also returns a non-nil error. TerminationOutcomeUnknown TerminationOutcome = iota // TerminationOutcomeSuccess means the request was transitioned from a non-terminal // state to the target terminal state and a terminal log entry was published. TerminationOutcomeSuccess // TerminationOutcomeAlreadyInTargetState means the request was already in the target terminal // state. No state write occurred, but the terminal log entry was re-published to // repair a possible prior attempt that wrote the state but failed before publishing. TerminationOutcomeAlreadyInTargetState // TerminationOutcomeDiverged means the request had already reached a different terminal state - // a concurrent path won the race and owns the terminal log for the state it wrote. // Nothing was written or published. TerminationOutcomeDiverged // TerminationOutcomeNotFound means the request does not exist. Nothing was done. TerminationOutcomeNotFound )
type TerminationResult ¶
type TerminationResult struct {
// Outcome is what happened to the request.
Outcome TerminationOutcome
// BeforeState is the request's state as observed before any write.
// On a divergence it is the (different) terminal state the request actually reached.
// On a success it is the prior non-terminal state.
// Empty when the request was not found or the call failed.
BeforeState entity.RequestState
// AfterState is the request's state after the operation.
// Equal to the target state on success and already-terminal.
// Equal to BeforeState on divergence.
// Empty when the request was not found or the call failed.
AfterState entity.RequestState
}
TerminationResult reports what TerminateRequest observed and did, so callers can log and emit metrics for the outcome at their own level and granularity without re-fetching the request. It is returned alongside a separate error: TerminationResult describes the expected variation (reconciled / already-terminal / diverged / not-found). Error signals an infrastructure failure.
func TerminateRequest ¶
func TerminateRequest( ctx context.Context, store storage.Storage, registry consumer.TopicRegistry, requestID string, targetState entity.RequestState, lastError string, metadata map[string]string, ) (TerminationResult, error)
TerminateRequest transitions a request to the given terminal state and publishes the corresponding terminal log entry, idempotently under at-least-once delivery. It is the shared primitive behind concluding a batch's requests, dead-letter reconciliation, and rejecting an invalid request at validation time.
targetState must be one of the terminal request states (Landed, Error, Cancelled). The write follows the immutability / optimistic-locking contract: caller-owned version arithmetic (newVersion = version+1) guards a pure conditional store write, and the in-memory version is only advanced after the store call succeeds.
Idempotency has three shapes, reported via TerminationResult.Outcome:
- the request is already in targetState: the state write is skipped but the terminal log is re-published (TerminationOutcomeAlreadyInTargetState);
- the request is in a different terminal state: nothing is written or published, since the other writer owns that state's terminal log (TerminationOutcomeDiverged);
- the request is not found: nothing is done (TerminationOutcomeNotFound).
The caller owns all logging and metrics — TerminationResult carries the state context needed to do so. lastError and metadata are attached to the published RequestLog for diagnosis.
A storage.ErrVersionMismatch from the conditional write is returned as-is (it is intrinsically retryable) so the caller's next attempt re-reads and re-evaluates.
Source Files