postgres

func (*Store) AbortBuildRetirement ¶

func (s *Store) AbortBuildRetirement(ctx context.Context, r durable.BuildRetirementRequest) (durable.LifecycleReceipt, error)

func (*Store) AbortQueryRuntimeRemoval ¶

func (s *Store) AbortQueryRuntimeRemoval(ctx context.Context, r durable.AbortQueryRemovalRequest) (durable.LifecycleReceipt, error)

func (*Store) AckEvent ¶

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

AckEvent acknowledges an event, marking it as consumed.

func (*Store) AcknowledgeDelivery ¶

func (s *Store) AcknowledgeDelivery(ctx context.Context, t durable.DeliveryToken, r durable.SinkReceipt) error

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 row-level locking to prevent race conditions.

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. Uses a multi-step approach: clear expired → check active → claim.

func (*Store) AppendSecurityAudit ¶

func (s *Store) AppendSecurityAudit(ctx context.Context, a durable.SecurityAudit) (durable.Delivery, error)

func (*Store) ApplyChildDelivery ¶

func (s *Store) ApplyChildDelivery(ctx context.Context, r durable.ChildDeliveryRequest) (result durable.ChildDeliveryReceipt, resultErr error)

ApplyChildDelivery serializes one target without locking the source execution.

func (*Store) ApplyExecutionTimeout ¶

func (s *Store) ApplyExecutionTimeout(ctx context.Context, r durable.ExecutionTimeoutRequest) (result durable.Receipt, resultErr error)

ApplyExecutionTimeout records closure and all lifecycle consequences atomically.

func (*Store) BeginBuildRetirement ¶

func (s *Store) BeginBuildRetirement(ctx context.Context, r durable.BuildRetirementRequest) (durable.LifecycleReceipt, error)

func (*Store) BeginLegacyAttempt ¶

func (s *Store) BeginLegacyAttempt(ctx context.Context, a durable.SecurityAudit) (durable.LegacyAttempt, error)

func (*Store) BeginQueryRuntimeRemoval ¶

func (s *Store) BeginQueryRuntimeRemoval(ctx context.Context, r durable.BeginQueryRemovalRequest) (durable.LifecycleReceipt, error)

func (*Store) BindQueryRemovalCommit ¶

func (s *Store) BindQueryRemovalCommit(ctx context.Context, r durable.QueryRemovalCommitRequest) (result durable.QueryRemovalCommitReceipt, resultErr error)

BindQueryRemovalCommit records an invocation only while current proof is fresh.

func (*Store) BlockDelivery ¶

func (s *Store) BlockDelivery(ctx context.Context, t durable.DeliveryToken) error

func (*Store) CheckDeliveryCompatibility ¶

func (s *Store) CheckDeliveryCompatibility(ctx context.Context, protocol int) error

func (*Store) CheckQueryRemovalCommit ¶

func (s *Store) CheckQueryRemovalCommit(ctx context.Context, q durable.QueryRemovalCommitLookup) (result durable.QueryRemovalCommitStatus, resultErr error)

CheckQueryRemovalCommit reports historical acceptance and exact current state.

func (*Store) CheckQueryRuntimeRemoval ¶

func (s *Store) CheckQueryRuntimeRemoval(ctx context.Context, f durable.QueryRemovalFence) (durable.QueryRemovalFacts, error)

func (*Store) ClaimChildDelivery ¶

func (s *Store) ClaimChildDelivery(ctx context.Context, r durable.ChildDeliveryClaimRequest) (*durable.ChildDelivery, error)

ClaimChildDelivery leases pending messages independently of source state.

func (*Store) ClaimDeliveries ¶

func (s *Store) ClaimDeliveries(ctx context.Context, r durable.DeliveryClaim) ([]durable.DeliveryRecord, error)

func (*Store) ClaimExecutionTimeout ¶

func (s *Store) ClaimExecutionTimeout(ctx context.Context, r durable.ExecutionTimeoutClaimRequest) (result *durable.ExecutionTimeoutTask, resultErr error)

ClaimExecutionTimeout skips owned rows and grants expiry across workflow builds.

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.

func (*Store) ClaimTask ¶

func (s *Store) ClaimTask(ctx context.Context, r durable.ClaimRequest) (result *durable.Task, resultErr error)

ClaimTask atomically claims one eligible task, skipping other pollers' locks.

func (*Store) ClaimTimeoutTask ¶

func (s *Store) ClaimTimeoutTask(ctx context.Context, r durable.TimeoutClaimRequest) (result *durable.Task, resultErr error)

ClaimTimeoutTask grants expired activity processing across queues. The timeout lease is independent of the expired business deadline and fences older grants.

func (*Store) Close ¶

func (s *Store) Close() error

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

func (*Store) CloseQueryRemovalGuard ¶

func (s *Store) CloseQueryRemovalGuard(ctx context.Context, r durable.QueryRemovalGuardCloseRequest) (result durable.QueryRemovalGuardCloseReceipt, resultErr error)

CloseQueryRemovalGuard retains the permanent cutoff and releases its pins.

func (*Store) CommitTransition ¶

func (s *Store) CommitTransition(ctx context.Context, r durable.CommitRequest) (result durable.Receipt, resultErr error)

CommitTransition serializes a run's mutations and fences task completion.

func (*Store) CompleteLegacyAttempt ¶

func (s *Store) CompleteLegacyAttempt(ctx context.Context, o durable.LegacyOutcome) error

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. A set FailedBefore counts entries that failed strictly before it, the same boundary PurgeDLQ deletes on.

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 in a single transaction.

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) DeferWorkflowTask ¶

func (s *Store) DeferWorkflowTask(ctx context.Context, r durable.WorkflowTaskDeferralRequest) (result durable.WorkflowTaskDeferralReceipt, resultErr error)

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 given threshold. Returns the number of rows deleted.

func (*Store) DeliveryStatus ¶

func (s *Store) DeliveryStatus(ctx context.Context, r durable.DeliveryStatusRequest) (durable.DeliveryStatus, error)

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.

The fit predicate is compiled into the same statement that performs the claim, so a job that does not fit is never written to: it stays pending and untouched for the next worker that does have room. The UPDATE ... WHERE id IN (SELECT ... FOR UPDATE SKIP LOCKED) shape that makes the claim atomic is unchanged; the predicate is simply another conjunct of the inner SELECT's WHERE.

When opts.Grants() the lease columns are additional assignments in that same UPDATE's SET clause, never a follow-up statement. ReclaimExpiredLeases requires lease_expires_at IS NOT NULL, so a row left running with a null expiry is not a row 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) EnrollRetirement ¶

func (s *Store) EnrollRetirement(ctx context.Context, r durable.RetirementEnrollmentRequest) (result durable.LifecycleReceipt, resultErr error)

func (*Store) FinalizeBuildRetirement ¶

func (s *Store) FinalizeBuildRetirement(ctx context.Context, r durable.BuildRetirementRequest) (durable.LifecycleReceipt, error)

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) FinishQueryRuntimeRemoval ¶

func (s *Store) FinishQueryRuntimeRemoval(ctx context.Context, r durable.FinishQueryRemovalRequest) (durable.LifecycleReceipt, error)

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) GetChildDelivery ¶

func (s *Store) GetChildDelivery(ctx context.Context, source durable.Key, id string) (durable.ChildDelivery, error)

GetChildDelivery reads the original routing and its current delivery state.

func (*Store) GetChildExecution ¶

func (s *Store) GetChildExecution(ctx context.Context, parent durable.Key, commandID string) (durable.ChildExecution, error)

GetChildExecution reads the saved relationship and the child's current state.

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. The dlq_replayed_job_id migration's (job_id, id) index answers it by reading one row from the end of that job's range.

func (*Store) GetExecution ¶

func (s *Store) GetExecution(ctx context.Context, key durable.Key) (durable.Execution, error)

GetExecution reads the persisted projection for one namespace-qualified run.

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) GetNamespace ¶

func (s *Store) GetNamespace(ctx context.Context, installation, namespace string) (durable.NamespaceRecord, error)

func (*Store) GetParentExecution ¶

func (s *Store) GetParentExecution(ctx context.Context, child durable.Key) (durable.ChildExecution, error)

GetParentExecution reads a child's unique original parent relationship.

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) GetTask ¶

func (s *Store) GetTask(ctx context.Context, key durable.Key, taskID string) (durable.Task, error)

GetTask reads pending or finished task state using its full execution identity.

func (*Store) GetWorker ¶

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

GetWorker returns one registered worker.

func (*Store) GetWorkflowTaskDeferral ¶

func (s *Store) GetWorkflowTaskDeferral(ctx context.Context, key durable.Key, taskID string) (durable.WorkflowTaskDeferral, error)

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) InspectBuildLifecycle ¶

func (s *Store) InspectBuildLifecycle(ctx context.Context, target durable.BuildTarget) (durable.BuildLifecycleFacts, error)

func (*Store) InspectCompatibility ¶

func (s *Store) InspectCompatibility(ctx context.Context, target durable.NamespaceTarget) (durable.CompatibilityFacts, error)

func (*Store) InspectQueryRetention ¶

func (s *Store) InspectQueryRetention(ctx context.Context, target durable.BuildTarget) (durable.QueryRetentionFacts, error)

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. Soft-deleted artifacts are left out unless IncludeDeleted is set.

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) ListChildDeliveries ¶

func (s *Store) ListChildDeliveries(ctx context.Context, source durable.Key, after string, limit int) ([]durable.ChildDelivery, error)

ListChildDeliveries pages by exclusive delivery ID, including completed rows.

func (*Store) ListChildExecutions ¶

func (s *Store) ListChildExecutions(ctx context.Context, parent durable.Key, after string, limit int) ([]durable.ChildExecution, error)

ListChildExecutions pages by command ID, with an exclusive cursor.

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) ListExecutions ¶

func (s *Store) ListExecutions(ctx context.Context, r durable.ExecutionList) ([]durable.Execution, string, error)

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) ListNamespaces ¶

func (s *Store) ListNamespaces(ctx context.Context, r durable.NamespaceList) ([]durable.NamespaceRecord, error)

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) ListQueryRuntimes ¶

func (s *Store) ListQueryRuntimes(ctx context.Context, r durable.QueryRuntimeList) (durable.QueryRuntimePage, error)

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) ListTasks ¶

func (s *Store) ListTasks(ctx context.Context, r durable.TaskList) ([]durable.Task, string, error)

func (*Store) ListWorkers ¶

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

ListWorkers returns all registered workers.

func (*Store) LookupLifecycleReceipt ¶

func (s *Store) LookupLifecycleReceipt(ctx context.Context, q durable.LifecycleReceiptLookup) (durable.LifecycleReceipt, error)

func (*Store) LookupQueryRemovalGuardReceipt ¶

func (s *Store) LookupQueryRemovalGuardReceipt(ctx context.Context, q durable.QueryRemovalGuardReceiptLookup) (durable.QueryRemovalGuardReceipt, error)

LookupQueryRemovalGuardReceipt performs exact nonissuing command recovery.

func (*Store) LookupQueryRemovalRevalidation ¶

func (s *Store) LookupQueryRemovalRevalidation(ctx context.Context, q durable.QueryRemovalRevalidationLookup) (durable.QueryRemovalRevalidationMaterial, error)

LookupQueryRemovalRevalidation recovers one immutable material row.

func (*Store) LookupReceipt ¶

func (s *Store) LookupReceipt(ctx context.Context, r durable.ReceiptRequest) (durable.Receipt, bool, error)

LookupReceipt reads one committed intent without locking or changing its run.

func (*Store) Migrate ¶

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

Migrate runs programmatic migrations via the grove orchestrator.

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. Links cascade.

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.

Postgres will not take a LIMIT on DELETE, so the rows are selected first. That is also what makes the bound meaningful: a retention sweep over a table this size has to be able to work in batches rather than locking the whole thing in one statement.

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 overwritten. The primary key is the only unique constraint on dispatch_dlq, so a unique violation here can only mean the ID.

func (*Store) ReadBuildFacts ¶

func (s *Store) ReadBuildFacts(ctx context.Context, namespace, build string) (durable.BuildFacts, error)

func (*Store) ReadDeliveryStatus ¶

func (s *Store) ReadDeliveryStatus(ctx context.Context, r durable.ScopedDeliveryStatus) (durable.DeliveryStatus, error)

func (*Store) ReadHistory ¶

func (s *Store) ReadHistory(ctx context.Context, key durable.Key, after int64, limit int) ([]durable.Event, error)

ReadHistory returns an ordered page without loading the entire history.

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 — or, for jobs whose worker died before the first heartbeat, whose start time — 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.

The claim and the read are one statement. The old select-then-update reaper let two pools both see the same stale job and both reset it.

func (*Store) RecordHeartbeat ¶

func (s *Store) RecordHeartbeat(ctx context.Context, r durable.HeartbeatRequest) (result durable.Receipt, resultErr error)

RecordHeartbeat commits task progress and its immutable request receipt.

func (*Store) RecordJobUsage ¶

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

RecordJobUsage persists one attempt's measurements.

func (*Store) RecordQueryRuntimeVerification ¶

func (s *Store) RecordQueryRuntimeVerification(ctx context.Context, r durable.VerifyQueryRuntimeRequest) (durable.LifecycleReceipt, error)

func (*Store) RegisterBuild ¶

func (s *Store) RegisterBuild(ctx context.Context, r durable.RegisterBuildRequest) (result durable.LifecycleReceipt, resultErr error)

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) RegisterNamespace ¶

func (s *Store) RegisterNamespace(ctx context.Context, c durable.NamespaceConfig) (durable.NamespaceRecord, error)

func (*Store) RegisterQueryRuntime ¶

func (s *Store) RegisterQueryRuntime(ctx context.Context, r durable.RegisterQueryRuntimeRequest) (durable.LifecycleReceipt, error)

func (*Store) RegisterWorker ¶

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

RegisterWorker adds a new worker to the cluster registry. Uses ON CONFLICT to upsert if the worker already exists.

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 undoes a claim, but only the one jobID made.

func (*Store) RenewDelivery ¶

func (s *Store) RenewDelivery(ctx context.Context, t durable.DeliveryToken, ttl time.Duration) (time.Time, error)

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) RenewTask ¶

func (s *Store) RenewTask(ctx context.Context, key durable.Key, token durable.TaskToken, ttl time.Duration) (result time.Time, resultErr error)

RenewTask extends a live grant without shortening its current deadline.

func (*Store) ReopenRun ¶

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

ReopenRun moves a finished run back to running.

func (*Store) ReplayDLQ ¶

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

ReplayDLQ marks a DLQ entry as replayed.

func (*Store) RequestCancelExecution ¶

func (s *Store) RequestCancelExecution(ctx context.Context, r durable.CancelExecutionRequest) (result durable.CancelExecutionReceipt, resultErr error)

RequestCancelExecution deduplicates before choosing the explicit or open run.

func (*Store) RequestWorkerDrain ¶

func (s *Store) RequestWorkerDrain(ctx context.Context, r durable.WorkerDrainRequest) (result durable.LifecycleReceipt, resultErr error)

func (*Store) ResolveExecution ¶

func (s *Store) ResolveExecution(ctx context.Context, r durable.ExecutionTarget) (durable.Execution, error)

ResolveExecution selects a projection using one database snapshot.

func (*Store) RetryDelivery ¶

func (s *Store) RetryDelivery(ctx context.Context, r durable.DeliveryRetry) error

func (*Store) RevalidateQueryRuntimeRemoval ¶

func (s *Store) RevalidateQueryRuntimeRemoval(ctx context.Context, r durable.QueryRemovalRevalidationRequest) (result durable.QueryRemovalRevalidationReceipt, resultErr error)

RevalidateQueryRuntimeRemoval retains fresh proof and durable pins atomically.

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.

func (*Store) SignalExecution ¶

func (s *Store) SignalExecution(ctx context.Context, r durable.SignalRequest) (result durable.SignalReceipt, resultErr error)

SignalExecution accepts a signal and workflow wakeup in one transaction.

func (*Store) SignalWithStart ¶

func (s *Store) SignalWithStart(ctx context.Context, r durable.SignalWithStartRequest) (durable.SignalReceipt, error)

SignalWithStart resolves its workflow-scoped receipt before choosing a run.

func (*Store) SignalWithStartOutcome ¶

func (s *Store) SignalWithStartOutcome(ctx context.Context, r durable.SignalWithStartRequest) (durable.SignalStartOutcome, error)

SignalWithStartOutcome reports receipt recovery from the acceptance transaction.

func (*Store) StartExecution ¶

func (s *Store) StartExecution(ctx context.Context, r durable.StartRequest) (result durable.Receipt, resultErr error)

StartExecution persists a run, its first event, task and receipt together.

func (*Store) StartWakeListener ¶

func (s *Store) StartWakeListener(ctx context.Context, wake func()) (func(), error)

StartWakeListener subscribes to the dispatch wake channel and invokes wake for each notification. The subscription holds one dedicated connection from the pool and is rebuilt with a delay if it dies (connection loss, failover). The returned stop function terminates the listener and blocks until its connection is released.

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 since grove is dialect-agnostic and LISTEN/NOTIFY is PostgreSQL-specific.

func (*Store) SweepEphemeral ¶

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

SweepEphemeral marks eligible ephemeral artifacts as deleted.

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) UnresolvedLegacyAttempts ¶

func (s *Store) UnresolvedLegacyAttempts(ctx context.Context, r durable.LegacyAttemptList) ([]durable.LegacyAttempt, error)

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 never 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.

func (*Store) UpdateRun ¶

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

UpdateRun persists changes to an existing workflow run.