Help us shape the future of Go! Take the 2026 Go Developer Survey

redis

package
v1.9.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 26, 2026 License: MIT Imports: 24 Imported by: 0

Documentation

Overview

Package redis implements store.Store using Grove KV backed by Redis. Suitable for high-throughput ephemeral workloads. Jobs use Sorted Sets as priority queues, events use Streams, and all entities are stored as JSON via Grove KV.

The caller owns the *kv.Store lifecycle -- redis never closes it. Pass the KV store through the constructor: 

import (
    "github.com/xraph/grove/kv"
    "github.com/xraph/dispatch/store/redis"
)

store := redis.New(kvStore)
if err := store.Ping(ctx); err != nil { ... }

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func MissingCapabilities 

func MissingCapabilities(store *kv.Store) []string

MissingCapabilities reports which of the kv capabilities this store needs the given driver does not provide. An empty result means the driver can back Dispatch.

Sorted sets order the queue, sets enumerate ids, hashes hold artifact links, scripts make the lease compare-and-set atomic, and streams carry events. Pub/Sub is absent from this list deliberately: it only shortens wake latency, and polling covers its absence.

Types

type Option 

type Option func(*Store)

Option configures the Store.

func WithKeyPrefix 

func WithKeyPrefix(prefix string) Option

WithKeyPrefix namespaces every key and channel this store touches so several dispatch instances can share one Redis without seeing each other's jobs, cron locks or leadership. Pass the tenant's own prefix (for example "ws_acme:"); the empty string keeps the historical keys.

func WithLogger 

func WithLogger(l log.Logger) Option

WithLogger sets a custom logger.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store implements the composite store.Store interface over Grove KV.

Every operation goes through kv.Store, including the sorted sets, hashes, scripts, and streams the job queue depends on. Nothing here imports a Redis client, so the backend this runs against is whichever kv driver the caller opened -- provided it supports those capabilities, which Store checks at construction.

func New 

func New(store *kv.Store, opts ...Option) *Store

New creates a new Redis KV-backed store. The caller owns the KV store lifecycle.

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. Of several workers racing for a free lock, exactly one gets true.

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.

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. The decision and the write are one compare-and-set (updateEntity), so of two concurrent claims exactly one wins and the other re-reads the entry, finds it claimed, and gets dispatch.ErrDLQAlreadyReplayed.

func (*Store) Close 

func (s *Store) Close() error

Close is a no-op -- the caller owns the KV store 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. It reads every entry: O(n) on Redis, like CountJobs.

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. It reads every run: O(n) on Redis, like CountJobs.

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.

The live-key guard is a SETNX on a dedicated key, which is what makes two concurrent creates at the same coordinates resolve to one winner and one ErrExists.

func (*Store) CreateRun 

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

CreateRun persists a new workflow run.

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 hard-deletes one entry together with every index that points at it, in one script. The entry is read first only to learn its job; whether this call deleted it is decided by the DEL inside the script, so of two concurrent deletes exactly one returns nil.

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 entries whose last-seen timestamp is older than the threshold. Returns the number of workers deleted.

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 call is a candidate scan followed by an exclusive claim:

  • dequeueCandidates reads the queue index, decodes each entity, applies job.DequeueOpts.Allows, sorts the SURVIVORS with job.DequeueOpts.Less and only then truncates to Limit. Order precedes truncation, over the whole eligible set — a scan that took Limit members first and sorted within them would hand a worker with a small limit arbitrary low-priority work forever, which is what storetest's LimitTruncatesAfterOrdering pins. How much of the index is read depends on whether the caller filters anything: see the unboundedScan* constants.
  • claimCandidates wins each survivor by removing it from the queue index, which is what makes the claim exclusive, then writes it back as running — carrying the lease grant when opts.Grants(), in that same write.

A job that does not fit is never removed from the index and never written to: it stays pending and untouched for the next worker that does have room. That is the whole point of the predicate — a job claimed and then requeued would bounce between small workers, delaying exactly the job that is hardest to place.

A non-positive Limit claims nothing, matching the SQL backends' LIMIT 0. A worker computing zero free slots must claim zero jobs, never the whole queue.

Empty opts.Queues claims nothing here, which is this backend's existing "all queues" behaviour: the queue index is one sorted set per queue name and there has never been a cross-queue index to scan. The conformance suite never exercises empty Queues, and no caller in this repository sends it.

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 stores the job as a JSON entity and adds it to the queue's Sorted Set.

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 CreatedAt descending — see the artifact.Store doc comment for why ties happen and what the tie-break does and does not fix.

In practice a tie cannot reach this loop on this backend: LinkArtifact stores each link in a hash keyed by linkField(name, attempt), so a second write to the same (owner, name, attempt) overwrites the first rather than coexisting as a second row the way the SQL and document backends allow. The CreatedAt compare is kept anyway, both so this backend agrees with the other four if that storage shape ever changes, and because ListLinks' own sort does not otherwise guarantee which of two equal-attempt entries — however they arose — comes first.

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.

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 the job's dlqByJob set, so it costs a few round trips however large the dead letter queue is, once ensureDLQJobIndex has nothing left to add.

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.

The read and the write are two round trips, so a DeleteStaleWorkers that lands between them deletes the entity and its worker_ids member, then the write puts the entity back alone. Re-adding the member here heals that: without it the heartbeat keeps succeeding against a worker ListWorkers cannot see, and the engine, which re-registers only on ErrWorkerNotFound, never notices. SAdd of an existing member is a no-op.

func (*Store) KV 

func (s *Store) KV() *kv.Store

KV returns the underlying KV store.

func (*Store) KeyPrefix 

func (s *Store) KeyPrefix() string

KeyPrefix returns the tenant prefix every key is written under; empty when the store uses the historical unprefixed keys.

func (*Store) LinkArtifact 

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

LinkArtifact records that an owner references an artifact, idempotently. HSet on the same field is naturally idempotent.

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.

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. See indexScan.page for when a page comes back incomplete.

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 (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(_ context.Context) error

Migrate is a no-op for Redis (schemaless).

func (*Store) Ping 

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

Ping verifies the Redis connection is alive.

func (*Store) PublishEvent 

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

PublishEvent persists a new event and adds it to the name's stream.

func (*Store) PurgeArtifact 

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

PurgeArtifact hard-deletes an artifact and its link index entries.

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.

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.

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, which would silently drop a replay claim. The SET NX and the index writes run in one script, so a refused push leaves no index member behind and an accepted one is never visible half-indexed.

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

Reclamation walks the job-id set rather than a sorted index, matching ReapStaleJobs — there is no secondary index of running-with-expired- lease jobs in this backend. Each candidate is filtered here in Go by reclaimable, the pending-state entity is computed in Go, and claimExpired does the compare-and-set: keyed on the epoch this call observed, so two pools scanning concurrently cannot both take the same job — whichever script call runs second sees an epoch (or state) that no longer matches and backs off.

func (*Store) RecordJobUsage 

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

RecordJobUsage persists one attempt's measurements.

Redis has no query engine, so the ordering and the two filters the contract requires have to come from indexes written here: a sorted set scored by record time for the global view and the retention sweep, and one per definition for the read an estimator actually makes.

func (*Store) RegisterCron 

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

RegisterCron persists a new cron entry.

func (*Store) RegisterWorker 

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

RegisterWorker adds a new worker to the cluster registry.

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. It is a no-op when the entry is gone or the lock is not this worker's.

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. On any other entry it writes nothing and returns nil.

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.

Go reads the current entity, mutates only the lease/heartbeat fields, and serializes the whole thing with encoding/json — the same path every other write in this store uses. The script's only job is the compare-and-set: verify state/worker_id/lease_epoch still match what this read saw, and if so, SET the blob Go built. See the file comment above for the ABA tradeoff this introduces and why it's accepted.

func (*Store) ReopenRun 

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

ReopenRun moves a finished run back to running. The decision and the write are one compare-and-set (updateEntity), so of two concurrent reopens exactly one wins and the other re-reads the run, finds it running, and is refused.

func (*Store) ReplayDLQ 

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

ReplayDLQ marks a DLQ entry as replayed. It goes through the same compare-and-set as ClaimReplay, so it can never overwrite a claim's replayed_job_id with a copy of the entry read before the claim.

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.

func (*Store) SetCronEnabled 

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

SetCronEnabled sets enabled, sets next_run_at when nextRunAt is non-nil, and stamps updated_at. Every other field keeps whatever the entry holds at the moment of the write.

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

The driver re-subscribes after connection loss, so there is no manual rebuild loop; messages published while the connection was down are simply lost, which polling covers. The returned stop function cancels the subscription.

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 stream polling for efficient waiting.

func (*Store) SweepEphemeral 

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

SweepEphemeral marks eligible ephemeral artifacts as deleted.

Candidates come from the ephemeral sorted set, which by construction contains no durable artifact. The lifecycle is re-checked after loading each candidate so the guard does not rest on index hygiene alone.

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.

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, and nothing else.

func (*Store) UpdateCronNextRun 

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

UpdateCronNextRun sets next_run_at and stamps updated_at, and never touches enabled.

func (*Store) UpdateJob 

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

UpdateJob persists changes to an existing job and keeps the queue index in step with the state it just wrote.

The index is what makes a job visible to DequeueJobs at all: EnqueueJob adds a member, the claim removes it, and every state transition after that arrives HERE. A retry from Runner.scheduleRetry, a job handed back by Pool.requeueRateLimited or Pool.requeueUndispatched, a stale job reset by Pool.reapStaleJobs — all of them just set a runnable state and call this. Writing that state without restoring the member leaves a job that GetJob reports as pending and no dequeue can ever see again: it looks healthy in the dashboard and runs nowhere. The other four backends have no equivalent hazard because they have no index — they re-derive candidacy from the row on every query.

The two index writes sit on OPPOSITE sides of the entity write, which is deliberate. The stored entity is authoritative — dequeue.go re-checks state and RunAt against it — so a member that should not be there is inert, while a member that is missing is a job that never runs again. Ordering each write so the index errs towards the harmless side means a lost second write cannot strand anything: 

becoming runnable — ZADD first, so a failed entity write leaves a
                    spare member the state gate ignores.
becoming final    — ZREM last, so a failed entity write leaves the job
                    both runnable and still indexed.

ZADD on a member already present only updates its score, so this is also safe for a job that was never claimed.

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.

This is the read-modify-write the column-exclusion shape forces: a whole-entity SET can't express "every field except these four," so Go reads the current entity, overlays only j's business fields onto it — never a pre-serialized blob built from j alone — and hands the result to renewLeaseScript for the same compare-and-set RenewLease uses. The fence the two need is identical (still running, still this worker, still this epoch), so reusing the script means the runner's terminal writes now go through the exact same compare-and-set family as renewal and reclaim. See the file comment above for exactly what that does and does not close.

lease_epoch, lease_expires_at, worker_id, and heartbeat_at are copied from the entity Go just read, never from j: j is the caller's stale snapshot, and every renewal since it was taken has pushed the real expiry forward. Writing j's copy of any of those back would roll the current holder's lease backwards even though the script's epoch check passes — see job.LeaseStore.UpdateLeasedJob.

It also mirrors UpdateJob's queue-index discipline (store/redis/job.go), which this method must not skip just because its own write is conditional. DequeueJobs claims a job with a ZPopMin off queueKey and never puts the member back on its own — see dequeue.go — so ANY write that lands a job in a runnable state (pending or retrying, the outcome of scheduleRetry) must restore the queue member, or the job becomes permanently unreachable: visible through GetJob, claimed by no future dequeue. handleSuccess and sendToDLQ happen to write terminal states, which never need the index touched at all, but that is a property of today's callers, not a license for this method to assume it.

func (*Store) UpdateRun 

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

UpdateRun persists changes to an existing workflow run.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.