Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewNoOpFlusher ¶
Types ¶
type ConcurrencyStrategy ¶
type ConcurrencyStrategy struct {
// contains filtered or unexported fields
}
func NewConcurrencyStrategy ¶
func NewConcurrencyStrategy( ctx context.Context, repo repository.ConcurrencyRepository, strategy *sqlcv1.V1StepConcurrency, outbox pgoutbox.Outbox, l *zerolog.Logger, ) *ConcurrencyStrategy
NewConcurrencyStrategy constructs a strategy index for a single (tenant, strategy) and registers it as the pgoutbox flusher for its topic (<tenant_id>.<strategy_id>). It kicks off the initial index hydration asynchronously on the provided (lifecycle) context, which must outlive any single Run - building can take much longer than a Run's deadline, and we must not abandon a partially-built index.
func (*ConcurrencyStrategy) Flush ¶
func (c *ConcurrencyStrategy) Flush(ctx pgoutbox.FlushContext, msgs []*outboxsqlc.Message) error
Flush satisfies the pgoutbox.Flusher interface. It runs inside the same transaction pgoutbox uses to acquire and delete the messages, so the slot writes performed here commit (or roll back) atomically with the message delete. We unmarshal the WAL payloads, replay them into the index, and stash the result for Run to collect. If we return an error, pgoutbox rolls the transaction back and the messages are redelivered on a later Run.
func (*ConcurrencyStrategy) Run ¶
func (c *ConcurrencyStrategy) Run(ctx context.Context) (*repository.RunConcurrencyResult, error)
Run drains the strategy's outbox topic, replaying every WAL message into the in-memory index and flushing the resulting slot decisions to the database. It returns the merged *repository.RunConcurrencyResult across all batches processed this tick.
Source Files