Method flush

enterprise/coderd/connectionlog/connectionlog.go:340–382

flush builds the batch params, clears the in-memory batch, and writes to the database. On failure, the batch is queued for retry by the single retry worker goroutine. If the retry queue is full, the batch is dropped.

(ctx context.Context)

// by the single retry worker goroutine. If the retry queue is full,
// the batch is dropped.
func (b *DBBatcher) flush(ctx context.Context) {
	count := b.batchLen()
	if count == 0 {
		return
	}

	params := b.buildParams()

	// Clear the batch before writing so the run loop can start
	// accumulating new entries.
	b.dedupedBatch = make(map[uuid.UUID]batchEntry, b.maxBatchSize)
	b.nullConnIDBatch = nil

	// Use the batcher's context for normal operation so Close()
	// can cancel hung writes. During shutdown (ctx already canceled),
	// fall back to a bounded timeout.
	writeCtx := b.ctx
	if writeCtx.Err() != nil {
		var cancel context.CancelFunc
		writeCtx, cancel = context.WithTimeout(context.Background(), shutdownWriteTimeout)
		defer cancel()
	}
	//nolint:gocritic // System-level batch operation for connection logs.
	err := b.store.BatchUpsertConnectionLogs(dbauthz.AsConnectionLogger(writeCtx), params)
	if err == nil {
		return
	}

	b.log.Error(ctx, "batch upsert failed, queueing for retry",
		slog.Error(err), slog.F("count", count))

	// Don't retry on shutdown.
	if ctx.Err() != nil {
		return
	}

	select {
	case b.retryCh <- params:
	default:
		b.log.Error(ctx, "retry queue full, dropping batch",
			slog.F("dropped", count))
	}
}

func (b *DBBatcher) buildParams() database.BatchUpsertConnectionLogsParams {
	count := b.batchLen()