Method Run

internal/queue/loop.go:26–43

Run blocks until ctx is cancelled. Each tick pings all enabled servers, pushes the batch, and (on success) drains up to DrainPerTick spooled batches in age order. On push failure the batch is written to the spool so it survives restarts.

(ctx context.Context)

// batches in age order. On push failure the batch is written to the spool
// so it survives restarts.
func (p *Producer) Run(ctx context.Context) {
	if existing, err := p.Spool.List(); err == nil && len(existing) > 0 {
		log.Warn("spool contains batches from previous run", "count", len(existing))
	}

	ticker := time.NewTicker(p.Interval)
	defer ticker.Stop()

	p.tick(ctx)
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			p.tick(ctx)
		}
	}
}

func (p *Producer) tick(ctx context.Context) {
	results := ping.QueryAll(p.Store.Snapshot())