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.
func (*ConcurrencyStrategy) UpdateStrategy ¶ added in v0.105.20
func (c *ConcurrencyStrategy) UpdateStrategy(next *sqlcv1.V1StepConcurrency)
UpdateStrategy applies a changed definition to the live index without a rebuild. The caller guarantees the strategy kind and parent linkage are unchanged (those alter the sub-queue comparators and heap ordering, so they require a rebuild); expression and static max-concurrency changes are safe to swap in place. buildingMu serializes this against WAL processing, index builds, and the queueing pass, so no batch observes a half-applied definition; mu covers getOrCreateSubQueue's reads.
The WAL path only re-decides sub-queues its messages touch, so a changed limit would otherwise not reach an idle group's backlog until new traffic arrived for it. Re-arm the all-sub-queue queueing pass (the same one that runs post-build) so the next Run applies the new limit everywhere: raises promote queued backlog immediately, lowers trim or grandfather per the strategy kind. Lock order matches runInitialQueueing (initialQueueMu before buildingMu).
Source Files