Method run

coderd/agentapi/metadatabatcher/metadata_batcher.go:264–297

run runs the batcher loop, reading from the update channel and flushing periodically or when the batch reaches capacity.

(ctx context.Context)

// run runs the batcher loop, reading from the update channel and flushing
// periodically or when the batch reaches capacity.
func (b *Batcher) run(ctx context.Context) {
	// nolint:gocritic // This is only ever used for one thing - updating agent metadata.
	authCtx := dbauthz.AsSystemRestricted(ctx)
	for {
		select {
		case update := <-b.updateCh:
			b.processUpdate(update)

			// Check if batch has reached capacity
			if int(b.currentBatchLen.Load()) >= b.maxBatchSize {
				b.flush(authCtx, flushCapacity)
				// Reset timer so the next scheduled flush is interval duration
				// from now, not from when it was originally scheduled.
				b.timer.Reset(b.interval, "metadataBatcher", "capacityFlush")
			}

		case <-b.timer.C:
			b.flush(authCtx, flushTicker)
			// Reset timer to schedule the next flush.
			b.timer.Reset(b.interval, "metadataBatcher", "scheduledFlush")

		case <-ctx.Done():
			b.log.Debug(ctx, "context done, flushing before exit")

			// We must create a new context here as the parent context is done.
			ctxTimeout, cancel := context.WithTimeout(context.Background(), finalFlushTimeout)
			defer cancel() //nolint:revive // We're returning, defer is fine.

			// nolint:gocritic // This is only ever used for one thing - updating agent metadata.
			b.flush(dbauthz.AsSystemRestricted(ctxTimeout), flushExit)
			return
		}
	}
}

// flush flushes the current batch to the database and pubsub.
func (b *Batcher) flush(ctx context.Context, reason string) {