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.
Source Files