command

func NewCompleteTaskExecutor ¶ added in v0.4.0

func NewCompleteTaskExecutor(
	client vaultapi.Client,
	vaultName string,
	currentDateTime libtime.CurrentDateTimeGetter,
	m metrics.Metrics,
) cdb.CommandObjectExecutorTx

NewCompleteTaskExecutor creates a cdb.CommandObjectExecutorTx that transitions a vault task to status: completed, phase: done with a # Resolution body section carrying the recovery SHA. The watcher publishes this on a build red→green transition (spec 076).

Resolution is by task_identifier (matching the CreateCommand's deterministic UUID5). Idempotent: a task that already carries a # Resolution block is a no-op (second completion does not append a duplicate block, matching the build-fix-close-obsolete-tasks skill's existing guard). vaultName is the controller's VAULT_NAME; a command whose non-empty TargetVault differs is cross-vault traffic and is skipped before the task read.

func NewCreateTaskExecutor ¶

func NewCreateTaskExecutor(
	gitClient gitclient.GitClient,
	client vaultapi.Client,
	taskDir string,
	vaultName string,
	currentDateTime libtime.CurrentDateTimeGetter,
	k int,
	resolver result.TaskPathResolver,
) cdb.CommandObjectExecutorTx

NewCreateTaskExecutor creates a cdb.CommandObjectExecutorTx that materializes a new vault task file for the given task_identifier. If cmd.Title passes validation the file is written at tasks/{title}.md; otherwise it falls back to tasks/{task_identifier}.md. A file that already exists at the resolved path causes ErrTaskAlreadyExists (a benign Failure on the result topic — no overwrite, no git write) UNLESS the existing file's status is terminal ("completed"/"aborted"), in which case a fresh non-terminal task is materialized at that path. Frontmatter must include "assignee" and "status"; missing fields return a wrapped validation error. Commands whose effective target vault (cmd.TargetVault or the legacy fallback) does not match vaultName are skipped without side effects (no git write, no error, no result event). resolver is the identifier→path index; a nil resolver records nothing and is a test-only value.

func NewExpiredCommandBatchHandler ¶ added in v0.19.0

func NewExpiredCommandBatchHandler(
	skip func(*sarama.ConsumerMessage) bool,
	observer ExpiredCommandObserver,
	handler libkafka.MessageHandlerBatch,
) libkafka.MessageHandlerBatch

NewExpiredCommandBatchHandler returns a batch handler that drops the messages skip reports and reports each of them to observer, then passes the remaining messages, in their original order, to handler in a single call.

It replaces libkafka.NewMessageHandlerBatchFilter rather than wrapping it, because that filter takes only the predicate and the handler and therefore has nowhere to report a drop. The partition-and-forward behaviour is otherwise identical, including the context check and the early return when every message in the batch was dropped.

skip must keep the contract libkafka states for it — pure, free of side effects, no state carried across calls — because the returned handler is shared by one consumer goroutine per partition. The observation deliberately does NOT live inside skip: a side effect there would run on every message of every partition concurrently and would couple the predicate's verdict to whatever the observer does. The wrapper observes after the verdict, per dropped message.

A dropped message is JSON-decoded a second time here, to hand the observer the command rather than only the raw bytes. That decode is paid only for dropped messages, which are rare, and it cannot fail in practice: skip returns true only after it decoded the same bytes successfully. The failure branch is kept so that a future predicate with different rules cannot silently swallow the observation.

func NewExpiredCommandSkip ¶ added in v0.18.4

func NewExpiredCommandSkip(
	commandExpireDuration time.Duration,
	currentDateTime libtime.CurrentDateTimeGetter,
) func(*sarama.ConsumerMessage) bool

NewExpiredCommandSkip returns a predicate reporting whether msg carries a command whose RequestTime is older than commandExpireDuration. True means the message is dropped; false means it is passed on.

The predicate exists to be composed outside cdb.CreateCommandMessageHandlerBatch. The handler already discards an expired command, but it does so from inside the batch transaction that NewMessageHandlerBatchTxUpdate opens, so the discard costs a bbolt write transaction and its fsync. Composed outside, the same command is dropped before the batch reaches the transaction.

The drop is logged at V(2) with the same wording the handler used, so moving the decision ahead of the transaction does not also silence it — an operator asking why a command never ran still gets the answer. That log is the predicate's only side effect: it carries no state, and glog is safe to call from the per-partition consumer goroutines that share this predicate.

A message that is not valid JSON is kept, so the handler reports the parse error exactly as it would without this filter. A payload that parses but carries no RequestTime is not kept: it unmarshals with a zero RequestTime, the comparison reads that as long expired, and the message is dropped — the same verdict the handler reaches, one step earlier.

A kept message is JSON-decoded twice, here and again inside cdb.NewCommandObjectMessageHandlerTx. That is deliberate: the saving is a whole batch transaction, and the decode is small beside it. Do not "optimise" the second decode away without removing this one too.

func NewIncrementFrontmatterExecutor ¶

func NewIncrementFrontmatterExecutor(
	client vaultapi.Client,
	vaultName string,
	m metrics.Metrics,
) cdb.CommandObjectExecutorTx

NewIncrementFrontmatterExecutor creates a cdb.CommandObjectExecutorTx that atomically reads the task file, increments the named frontmatter field by delta, and commits. If trigger_count reaches an explicitly-set max_triggers the assignee is cleared and phase is preserved in the same write. An absent max_triggers means no cap: the routing assignee is never stripped on recurring tasks that accumulate trigger_count. vaultName is the controller's VAULT_NAME; a command whose non-empty TargetVault differs is cross-vault traffic and is skipped before the task-file lookup. resolver is the scanner's identifier→path index, consulted by the lookup before it walks the vault.

func NewTaskResultExecutor ¶

func NewTaskResultExecutor(
	writer result.ResultWriter,
	retryGate PlanningRetryGate,
	vaultName string,
) cdb.CommandObjectExecutorTx

NewTaskResultExecutor creates a cdb.CommandObjectExecutorTx for update commands. Uses cdb.CommandObjectExecutorTxFunc adapter — same pattern as trading command handlers. vaultName is the controller's VAULT_NAME; a result whose target_vault frontmatter (stamped at task create, echoed by the agent) does not match is cross-vault traffic and is skipped before the write scan (no write, no not_found count, no result event).

func NewUpdateFrontmatterExecutor ¶

func NewUpdateFrontmatterExecutor(
	client vaultapi.Client,
	vaultName string,
	m metrics.Metrics,
) cdb.CommandObjectExecutorTx

NewUpdateFrontmatterExecutor creates a cdb.CommandObjectExecutorTx that reads the task through the typed vault API by identifier, PATCHes only the frontmatter fields the command names (plus the two doctrines), and PUTs the body when the command carries a body section. All other frontmatter keys are left untouched — the field level PATCH is what guarantees that, so no whole-file merge happens and none of the task's frontmatter is rewritten.

It is explicitly NOT atomic: PATCH and PUT are two requests and the typed surface has no combined primitive, so a failure between them leaves frontmatter updated and the body stale. See applyUpdate for the window and why the retry is safe.

vaultName is the controller's VAULT_NAME; a command whose non-empty TargetVault differs is cross-vault traffic and is skipped before the task lookup (no write, no counter, no result event).