factory

func CreateCommandConsumer ¶

func CreateCommandConsumer(
	saramaClientProvider libkafka.SaramaClientProvider,
	syncProducer libkafka.SyncProducer,
	db libkv.DB,
	batchSize libkafka.BatchSize,
	topicPrefix base.TopicPrefix,
	resultWriter result.ResultWriter,
	resolver result.TaskPathResolver,
	gitClient gitclient.GitClient,
	taskDir string,
	vaultName string,
	currentDateTime libtime.CurrentDateTimeGetter,
	k int,
	prCommenter prcomment.PRCommenter,
	m metrics.Metrics,
) run.Func

CreateCommandConsumer wires a CQRS command consumer for agent-task-v1-request.

batchSize bounds how many messages share one libkv.Tx. The handler composition (cdb.CreateCommandMessageHandlerBatch) is already MessageHandlerBatchTx — one db.Update per batch — so a batch size of 1 makes every message pay its own BoltDB transaction and fsync. Raising it is the only lever that changes the transaction count; the handler choice itself is already the batch variant.

func RetryConsumer ¶ added in v0.17.0

func RetryConsumer(consumer run.Func, delay time.Duration) run.Func

RetryConsumer wraps a command consumer so a transient Kafka failure reconnects instead of terminating the process.

The consumer is expected to run for the life of the pod, so any return is a transport-level interruption — `kafka: error while consuming <topic>/<partition>: EOF`, `broker not connected` — and not a reason to exit. Propagating it reaches service.Run, which treats the error as fatal: the controller stops scanning and publishing until Kubernetes restarts it, every in-flight command waits out the restart, and the request topic accumulates lag that is then drained one message at a time. Measured 2026-10-22: six restarts in 5h31m, each preceded by a consume EOF at a lag of 257–265.

The retry stops only when ctx is done, so a shutdown is a clean nil rather than an error the caller would have to special-case.