Method run

lineage_worker.go:145–161

run is the main processing loop.

(ctx context.Context)

// run is the main processing loop.
func (p *LineageOutboxProcessor) run(ctx context.Context) {
	ticker := time.NewTicker(p.pollInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			logrus.Info("Lineage outbox processor context cancelled")
			return
		case <-p.stopCh:
			logrus.Info("Lineage outbox processor stop signal received")
			return
		case <-ticker.C:
			p.processBatch(ctx)
		}
	}
}

// processBatch claims and processes a batch of pending outbox entries.
func (p *LineageOutboxProcessor) processBatch(ctx context.Context) {