Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func PublishBatchLogs ¶
func PublishBatchLogs(ctx context.Context, registry consumer.TopicRegistry, requestIDs []string, status entity.RequestStatus, metadata map[string]string) error
PublishBatchLogs publishes a request 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.
func PublishLog ¶
func PublishLog(ctx context.Context, registry consumer.TopicRegistry, logEntry entity.RequestLog, partitionKey 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) 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).
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.
func NewMaterializer ¶
func NewMaterializer(store storage.Storage) *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