mongo

func (*Store) AckEvent ¶

func (s *Store) AckEvent(ctx context.Context, eventID id.EventID) error

AckEvent acknowledges an event, marking it as consumed.

func (*Store) AcquireCronLock ¶

func (s *Store) AcquireCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID, ttl time.Duration) (bool, error)

AcquireCronLock attempts to acquire a distributed lock for a cron entry. Uses FindOneAndUpdate for atomic lock acquisition.

func (*Store) AcquireLeadership ¶

func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error)

AcquireLeadership attempts to become the cluster leader.

Relies on the partial unique index on is_leader=true (created by Migrate) so the index itself enforces single-leader. Two round-trips: (1) clear expired leaders, (2) atomic UpdateOne to claim. A duplicate-key error on step 2 means another worker already holds leadership.

func (*Store) ClaimReplay ¶

func (s *Store) ClaimReplay(ctx context.Context, entryID id.DLQID, jobID id.JobID) error

ClaimReplay marks an unreplayed entry replayed by jobID.

"Unreplayed" is replayed_at equal to nil, which matches both shapes an unreplayed entry can have: the explicit null grove's insert writes, and the absent key a raw-driver write or ReleaseReplay leaves.

func (*Store) Close ¶

func (s *Store) Close() error

Close is a no-op because the caller owns the *grove.DB lifecycle.

func (*Store) CountDLQ ¶

func (s *Store) CountDLQ(ctx context.Context) (int64, error)

CountDLQ returns the total number of entries in the dead letter queue.

func (*Store) CountDLQEntries ¶

func (s *Store) CountDLQEntries(ctx context.Context, opts dlq.CountOpts) (int64, error)

CountDLQEntries counts dead letter entries under the given filters. FailedBefore is strictly before, the same $lt PurgeDLQ deletes on, so the count a purge confirmation shows is the count the purge removes.

func (*Store) CountJobs ¶

func (s *Store) CountJobs(ctx context.Context, opts job.CountOpts) (int64, error)

CountJobs returns the number of jobs matching the given options.

func (*Store) CountRuns ¶

func (s *Store) CountRuns(ctx context.Context, opts workflow.CountRunsOpts) (int64, error)

CountRuns counts workflow runs by state and exact name.

func (*Store) CreateArtifact ¶

func (s *Store) CreateArtifact(ctx context.Context, a *artifact.Artifact, link *artifact.Link) error

CreateArtifact inserts an artifact and, when link is non-nil, its first link.

Mongo multi-document atomicity needs a replica set. When the deployment supports transactions the pair is written in a session; otherwise the artifact is written first and the link second, and a crash between them leaves an orphan that the orphan sweep collects.

func (*Store) CreateRun ¶

func (s *Store) CreateRun(ctx context.Context, run *workflow.Run) error

CreateRun persists a new workflow run.

func (*Store) DB ¶

func (s *Store) DB() *grove.DB

DB returns the underlying *grove.DB for advanced usage.

func (*Store) DeleteCheckpointsAfter ¶

func (s *Store) DeleteCheckpointsAfter(ctx context.Context, runID id.RunID, afterStep string) error

DeleteCheckpointsAfter removes all checkpoints created after the given step name (by creation order). Used for workflow replay.

func (*Store) DeleteCron ¶

func (s *Store) DeleteCron(ctx context.Context, entryID id.CronID) error

DeleteCron removes a cron entry by ID.

func (*Store) DeleteDLQ ¶

func (s *Store) DeleteDLQ(ctx context.Context, entryID id.DLQID) error

DeleteDLQ removes one entry.

func (*Store) DeleteJob ¶

func (s *Store) DeleteJob(ctx context.Context, jobID id.JobID) error

DeleteJob removes a job by ID.

func (*Store) DeleteStaleWorkers ¶

func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration) (int64, error)

DeleteStaleWorkers removes worker rows whose last-seen timestamp is older than the threshold. Used at startup to evict rows from previous instances that crashed before deregistering. Without this, a stale row with is_leader=true blocks the partial-unique leader index and the new instance can't claim leadership until mongo's TTL sweeper catches up.

func (*Store) DequeueJobs ¶

func (s *Store) DequeueJobs(ctx context.Context, opts job.DequeueOpts) ([]*job.Job, error)

DequeueJobs atomically claims up to opts.Limit ready jobs from opts.Queues that fit opts, sets them to running, and returns them ordered by priority descending, then locality-preferred first, then RunAt ascending.

Mongo has no statement that can order, limit, and claim many documents in one shot, so the claim is composed of two parts:

• one ordered candidate read, which applies the fit predicate, the full contract ordering, and the limit — order THEN truncate, server-side, over the whole eligible set;
• one FindOneAndUpdate per candidate, keyed by _id and carrying the SAME fit predicate plus the state and run_at guards. findAndModify is atomic per document, so two workers racing for one job produce exactly one winner; the loser's call matches nothing and it simply gets no job, never a second claim of the same one.

The read is also the write gate the previous probe provided: when it finds no candidate, not a single write command is sent. A job that does not fit is never written to — the predicate is a conjunct of the claiming update itself, not a filter over claimed documents.

When opts.Grants() the lease fields are part of that same per-document update, never a follow-up write. Per-document atomicity is what makes the claim exclusive, and it is also what keeps the grant recoverable: ReclaimExpiredLeases matches lease_expires_at {$ne: nil}, so a document left running with no expiry is not one at risk of reclamation — it is one reclamation can never see. See job.DequeueOpts.LeaseUntil.

func (*Store) DeregisterWorker ¶

func (s *Store) DeregisterWorker(ctx context.Context, workerID id.WorkerID) error

DeregisterWorker removes a worker from the cluster registry.

func (*Store) EnqueueJob ¶

func (s *Store) EnqueueJob(ctx context.Context, j *job.Job) error

EnqueueJob persists a new job in pending state.

func (*Store) FindArtifactByKey ¶

func (s *Store) FindArtifactByKey(ctx context.Context, backend, bucket, key string) (*artifact.Artifact, error)

FindArtifactByKey retrieves a live artifact by its storage coordinates.

func (*Store) FindLinkByName ¶

func (s *Store) FindLinkByName(
	ctx context.Context,
	owner artifact.OwnerRef,
	name string,
) (*artifact.Link, error)

FindLinkByName returns the highest-attempt link for an owner and name, breaking ties by created_at descending — see the artifact.Store doc comment for why ties happen and what the tie-break does and does not fix.

func (*Store) GetArtifact ¶

func (s *Store) GetArtifact(ctx context.Context, artifactID id.ArtifactID) (*artifact.Artifact, error)

GetArtifact retrieves a live artifact by ID.

func (*Store) GetArtifactRecord ¶

func (s *Store) GetArtifactRecord(ctx context.Context, artifactID id.ArtifactID) (*artifact.Artifact, error)

GetArtifactRecord returns metadata even after soft deletion.

func (*Store) GetCheckpoint ¶

func (s *Store) GetCheckpoint(ctx context.Context, runID id.RunID, stepName string) ([]byte, error)

GetCheckpoint retrieves checkpoint data for a specific workflow step. Returns nil data if no checkpoint exists.

func (*Store) GetCron ¶

func (s *Store) GetCron(ctx context.Context, entryID id.CronID) (*cron.Entry, error)

GetCron retrieves a cron entry by ID.

func (*Store) GetDLQ ¶

func (s *Store) GetDLQ(ctx context.Context, entryID id.DLQID) (*dlq.Entry, error)

GetDLQ retrieves a DLQ entry by ID.

func (*Store) GetDLQByJobID ¶

func (s *Store) GetDLQByJobID(ctx context.Context, jobID id.JobID) (*dlq.Entry, error)

GetDLQByJobID returns the newest entry, by ID, for a failed job. It reads through the {job_id: 1, _id: -1} index Migrate creates, so the first document the index yields is the answer.

func (*Store) GetJob ¶

func (s *Store) GetJob(ctx context.Context, jobID id.JobID) (*job.Job, error)

GetJob retrieves a job by ID.

func (*Store) GetLeader ¶

func (s *Store) GetLeader(ctx context.Context) (*cluster.Worker, error)

GetLeader returns the current cluster leader, or nil if there is no leader.

func (*Store) GetRun ¶

func (s *Store) GetRun(ctx context.Context, runID id.RunID) (*workflow.Run, error)

GetRun retrieves a workflow run by ID.

func (*Store) GetWorker ¶

func (s *Store) GetWorker(ctx context.Context, workerID id.WorkerID) (*cluster.Worker, error)

GetWorker returns one registered worker.

func (*Store) HeartbeatJob ¶

func (s *Store) HeartbeatJob(ctx context.Context, jobID id.JobID, _ id.WorkerID) error

HeartbeatJob updates the heartbeat timestamp for a running job.

func (*Store) HeartbeatWorker ¶

func (s *Store) HeartbeatWorker(ctx context.Context, workerID id.WorkerID) error

HeartbeatWorker updates the last-seen timestamp for a worker.

func (*Store) HeartbeatWorkers ¶

func (s *Store) HeartbeatWorkers(ctx context.Context, workerIDs []id.WorkerID) error

HeartbeatWorkers updates last-seen for many workers in one round-trip via BulkWrite. This is an opt-in fast path callers can reach via type assertion; the cluster.Store interface only requires HeartbeatWorker.

func (*Store) LinkArtifact ¶

func (s *Store) LinkArtifact(ctx context.Context, link *artifact.Link) error

LinkArtifact records that an owner references an artifact, idempotently.

func (*Store) ListArtifacts ¶

func (s *Store) ListArtifacts(ctx context.Context, opts artifact.ListOpts) ([]*artifact.Artifact, error)

ListArtifacts returns artifacts matching the given options, newest first.

func (*Store) ListArtifactsByOwner ¶

func (s *Store) ListArtifactsByOwner(
	ctx context.Context,
	owner artifact.OwnerRef,
	role artifact.Role,
) ([]*artifact.Artifact, error)

ListArtifactsByOwner returns live artifacts linked to an owner.

func (*Store) ListArtifactsPage ¶

func (s *Store) ListArtifactsPage(ctx context.Context, opts artifact.PageOpts) (artifact.Page, error)

ListArtifactsPage returns artifacts newest first by ID, filtered and paged. A live artifact's deleted_at may be absent or null, as in ListArtifacts, and equality with nil matches both.

func (*Store) ListCheckpoints ¶

func (s *Store) ListCheckpoints(ctx context.Context, runID id.RunID) ([]*workflow.Checkpoint, error)

ListCheckpoints returns all checkpoints for a workflow run.

func (*Store) ListChildRuns ¶

func (s *Store) ListChildRuns(ctx context.Context, parentRunID id.RunID) ([]*workflow.Run, error)

ListChildRuns returns all child workflow runs for a parent.

func (*Store) ListCrons ¶

func (s *Store) ListCrons(ctx context.Context) ([]*cron.Entry, error)

ListCrons returns all cron entries.

func (*Store) ListDLQ ¶

func (s *Store) ListDLQ(ctx context.Context, opts dlq.ListOpts) ([]*dlq.Entry, error)

ListDLQ returns DLQ entries matching the given options.

func (*Store) ListDLQPage ¶

func (s *Store) ListDLQPage(ctx context.Context, opts dlq.PageOpts) (dlq.Page, error)

ListDLQPage returns dead letter entries newest first by ID, filtered and paged.

func (*Store) ListJobUsage ¶

func (s *Store) ListJobUsage(ctx context.Context, opts job.UsageListOpts) ([]*job.Usage, error)

ListJobUsage returns recorded attempts, most recent first.

func (*Store) ListJobs ¶

func (s *Store) ListJobs(ctx context.Context, opts job.ListJobsOpts) (job.Page, error)

ListJobs returns jobs newest first by ID, filtered and paged.

func (*Store) ListJobsByState ¶

func (s *Store) ListJobsByState(ctx context.Context, state job.State, opts job.ListOpts) ([]*job.Job, error)

ListJobsByState returns jobs matching the given state.

func (*Store) ListLinks ¶

func (s *Store) ListLinks(ctx context.Context, owner artifact.OwnerRef) ([]*artifact.Link, error)

ListLinks returns every link belonging to the given owner.

func (*Store) ListPurgeable ¶

func (s *Store) ListPurgeable(
	ctx context.Context,
	grace time.Duration,
	limit int,
) ([]*artifact.Artifact, error)

ListPurgeable returns soft-deleted artifacts older than grace.

func (*Store) ListRuns ¶

func (s *Store) ListRuns(ctx context.Context, opts workflow.ListOpts) ([]*workflow.Run, error)

ListRuns returns workflow runs matching the given options.

func (*Store) ListRunsPage ¶

func (s *Store) ListRunsPage(ctx context.Context, opts workflow.ListRunsPageOpts) (workflow.RunPage, error)

ListRunsPage returns workflow runs newest first by ID, filtered and paged.

func (*Store) ListWorkers ¶

func (s *Store) ListWorkers(ctx context.Context) ([]*cluster.Worker, error)

ListWorkers returns all registered workers.

func (*Store) Migrate ¶

func (s *Store) Migrate(ctx context.Context) error

Migrate creates indexes for all dispatch collections.

CreateMany is itself idempotent — mongo silently no-ops indexes that already exist with matching specs — so this is safe to call on every boot.

Note this is the whole of the mongo backend's migration path: the grove migration group in migrations.go is not run from anywhere, so anything that must happen on upgrade belongs here rather than there. Jobs left running by a pre-lease build are deliberately NOT handled here. A one-shot backfill cannot see a job an old pod claims after the migration has already run, and it would evict jobs those pods are still running, because it seeds an expiry in the past that their heartbeats do not know to push. ReclaimExpiredLeases adopts them instead, continuously and gated on silence — see job.UnleasedReclaimGrace.

func (*Store) Ping ¶

func (s *Store) Ping(ctx context.Context) error

Ping checks database connectivity.

func (*Store) PublishEvent ¶

func (s *Store) PublishEvent(ctx context.Context, evt *event.Event) error

PublishEvent persists a new event and makes it available for subscribers.

func (*Store) PurgeArtifact ¶

func (s *Store) PurgeArtifact(ctx context.Context, artifactID id.ArtifactID) error

PurgeArtifact hard-deletes an artifact and its links.

func (*Store) PurgeDLQ ¶

func (s *Store) PurgeDLQ(ctx context.Context, before time.Time) (int64, error)

PurgeDLQ removes DLQ entries with FailedAt before the given time. Returns the number of entries removed.

func (*Store) PurgeJobUsage ¶

func (s *Store) PurgeJobUsage(ctx context.Context, before time.Time, limit int) (int64, error)

PurgeJobUsage deletes records older than before, up to limit rows.

DeleteMany takes no limit, so a bounded purge selects its victims first. An unbounded one deletes in a single statement, which is what the caller asked for by passing zero.

func (*Store) PushDLQ ¶

func (s *Store) PushDLQ(ctx context.Context, entry *dlq.Entry) error

PushDLQ adds a failed job entry to the dead letter queue. An ID that is already there is refused with dispatch.ErrDLQAlreadyExists rather than reported as a driver error: the _id index refuses the insert, so a push can never overwrite an entry or drop its replay claim.

func (*Store) ReapDeadWorkers ¶

func (s *Store) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([]*cluster.Worker, error)

ReapDeadWorkers returns workers whose last-seen timestamp is older than the given threshold.

func (*Store) ReapStaleJobs ¶

func (s *Store) ReapStaleJobs(ctx context.Context, threshold time.Duration) ([]*job.Job, error)

ReapStaleJobs returns running jobs whose last heartbeat is older than the given threshold.

func (*Store) ReclaimExpiredLeases ¶

func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job, error)

ReclaimExpiredLeases returns expired-lease jobs to pending, fencing their previous holders.

Each job is claimed by its own conditional FindOneAndUpdate keyed on the epoch it was seen at, so two pools reclaiming concurrently cannot both take the same job — the loser's filter no longer matches.

func (*Store) RecordJobUsage ¶

func (s *Store) RecordJobUsage(ctx context.Context, u *job.Usage) error

RecordJobUsage persists one attempt's measurements.

func (*Store) RegisterCron ¶

func (s *Store) RegisterCron(ctx context.Context, entry *cron.Entry) error

RegisterCron persists a new cron entry. Returns an error if the name already exists.

func (*Store) RegisterWorker ¶

func (s *Store) RegisterWorker(ctx context.Context, w *cluster.Worker) error

RegisterWorker adds a new worker to the cluster registry. Uses upsert to handle re-registration.

func (*Store) ReleaseCronLock ¶

func (s *Store) ReleaseCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID) error

ReleaseCronLock releases the distributed lock for a cron entry.

func (*Store) ReleaseReplay ¶

func (s *Store) ReleaseReplay(ctx context.Context, entryID id.DLQID, jobID id.JobID) error

ReleaseReplay clears a claim, but only the one jobID made. The filter names the job, so a release that lost to another claim matches nothing and changes nothing.

func (*Store) RenewLeadership ¶

func (s *Store) RenewLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error)

RenewLeadership extends the leader's hold.

func (*Store) RenewLease ¶

func (s *Store) RenewLease(
	ctx context.Context,
	jobID id.JobID,
	workerID id.WorkerID,
	epoch int,
	leaseUntil time.Time,
) error

RenewLease extends the lease only if the caller still holds it.

func (*Store) ReopenRun ¶

func (s *Store) ReopenRun(ctx context.Context, runID id.RunID, expectedGeneration int64) error

ReopenRun moves a run that is not running back to running, clearing its error and completion time. The state condition is in the filter, so of two concurrent reopens exactly one matches.

func (*Store) ReplayDLQ ¶

func (s *Store) ReplayDLQ(ctx context.Context, entryID id.DLQID) error

ReplayDLQ marks a DLQ entry as replayed.

func (*Store) SaveCheckpoint ¶

func (s *Store) SaveCheckpoint(ctx context.Context, runID id.RunID, stepName string, data []byte) error

SaveCheckpoint persists checkpoint data for a workflow step. If a checkpoint already exists for the same run/step, it is replaced.

func (*Store) SetCronEnabled ¶

func (s *Store) SetCronEnabled(ctx context.Context, entryID id.CronID, enabled bool, nextRunAt *time.Time) error

SetCronEnabled sets enabled, and next_run_at when one is given. It writes only those fields and updated_at, never the whole document.

func (*Store) SubscribeEvent ¶

func (s *Store) SubscribeEvent(ctx context.Context, name string, timeout time.Duration) (*event.Event, error)

SubscribeEvent waits for an unacked event matching the given name. Uses a polling approach with short intervals.

func (*Store) SweepEphemeral ¶

func (s *Store) SweepEphemeral(
	ctx context.Context,
	opts artifact.SweepOpts,
) ([]*artifact.Artifact, error)

SweepEphemeral marks eligible ephemeral artifacts as deleted.

Mongo cannot join links to jobs and runs in a single expressive statement the way SQL can, so eligibility is computed in two steps: candidates are narrowed by the ephemeral guard, then each candidate's owners are resolved and checked. The lifecycle guard is applied both when selecting candidates and again on the write.

func (*Store) SweepOrphans ¶

func (s *Store) SweepOrphans(
	ctx context.Context,
	cutoff time.Time,
	limit int,
) ([]*artifact.Artifact, error)

SweepOrphans marks link-less ephemeral artifacts created before cutoff.

func (*Store) UpdateArtifact ¶

func (s *Store) UpdateArtifact(ctx context.Context, a *artifact.Artifact) error

UpdateArtifact persists size, hash, content type, and expiry. Lifecycle, created_at, and deleted_at are deliberately not updatable here.

func (*Store) UpdateCronEntry ¶

func (s *Store) UpdateCronEntry(ctx context.Context, entry *cron.Entry) error

UpdateCronEntry updates a cron entry (Enabled, NextRunAt, etc.).

func (*Store) UpdateCronLastRun ¶

func (s *Store) UpdateCronLastRun(ctx context.Context, entryID id.CronID, at time.Time) error

UpdateCronLastRun records when a cron entry last fired.

func (*Store) UpdateCronNextRun ¶

func (s *Store) UpdateCronNextRun(ctx context.Context, entryID id.CronID, nextRunAt time.Time) error

UpdateCronNextRun sets next_run_at and updated_at, and nothing else, so the scheduler's write after a fire can never touch enabled.

func (*Store) UpdateJob ¶

func (s *Store) UpdateJob(ctx context.Context, j *job.Job) error

UpdateJob persists changes to an existing job.

func (*Store) UpdateLeasedJob ¶

func (s *Store) UpdateLeasedJob(ctx context.Context, j *job.Job, workerID id.WorkerID, epoch int) error

UpdateLeasedJob persists j only while the caller still holds the lease.

Unlike UpdateJob's ReplaceOne, this is an UpdateOne with the fence predicate IN THE FILTER and a $set of named business fields — no read-modify-write needed, and the document is never replaced wholesale. worker_id, lease_epoch, lease_expires_at, and heartbeat_at are never assigned in $set: j is the caller's stale snapshot, so writing its copy of lease_expires_at back would roll the current holder's expiry backwards even though the filter's epoch check passed.

resource_requests and resource_limits mirror toJobModel's zero-Set handling: a zero Set is $unset rather than $set to an empty document, matching the "absent key, not an empty subdocument" contract UpdateJob keeps via bson "omitempty" on ReplaceOne.

func (*Store) UpdateRun ¶

func (s *Store) UpdateRun(ctx context.Context, run *workflow.Run) error

UpdateRun persists changes to an existing workflow run.