From b17fb95a2bcf0e9cb95223a1c6f697e3050af9d7 Mon Sep 17 00:00:00 2001 From: Rex Raphael Date: Mon, 14 Sep 2026 15:54:41 -0500 Subject: [PATCH] feat(store/redis): namespace keys per tenant with WithKeyPrefix If you run more than one dispatch instance against the same Redis, they all write under dispatch:* and start dequeuing each other's jobs, treating each other's cron names as duplicates and fighting over the leader key. Nothing outside the store can fix that: streams, the wake channel and the lease scripts go straight to go-redis, so a grove namespace hook never sees them. Every key the store touches now comes from one keys type on the Store. redis.WithKeyPrefix("ws_acme:") puts the tenant segment outside the dispatch namespace (ws_acme:dispatch:job:), the same layout a tenant's other Redis keys use, so one SCAN pattern covers everything it owns. An empty prefix is byte for byte the old key, so existing deployments keep their data. The forge extension reads it as key_prefix in YAML or WithKVKeyPrefix from Go and passes it through on the grove KV path only. Tests cover job, cron and leadership isolation between two prefixed stores on one container, the legacy key shape for an empty prefix, and YAML winning the merge. --- docs/content/docs/guides/forge-extension.mdx | 16 ++ docs/content/docs/stores/redis.mdx | 16 ++ extension/config.go | 6 + extension/extension.go | 12 +- extension/kvprefix_internal_test.go | 66 ++++++++ extension/options.go | 9 + store/redis/artifact.go | 60 +++---- store/redis/cluster.go | 48 +++--- store/redis/cron.go | 28 ++-- store/redis/dequeue.go | 8 +- store/redis/dlq.go | 20 +-- store/redis/event.go | 10 +- store/redis/job.go | 32 ++-- store/redis/keys.go | 132 ++++++++------- store/redis/keys_internal_test.go | 23 +++ store/redis/lease.go | 14 +- store/redis/prefix_test.go | 165 +++++++++++++++++++ store/redis/store.go | 14 ++ store/redis/store_test.go | 16 +- store/redis/wake.go | 8 +- store/redis/workflow.go | 30 ++-- 21 files changed, 540 insertions(+), 193 deletions(-) create mode 100644 extension/kvprefix_internal_test.go create mode 100644 store/redis/keys_internal_test.go create mode 100644 store/redis/prefix_test.go diff --git a/docs/content/docs/guides/forge-extension.mdx b/docs/content/docs/guides/forge-extension.mdx index 9771805..916864d 100644 --- a/docs/content/docs/guides/forge-extension.mdx +++ b/docs/content/docs/guides/forge-extension.mdx @@ -47,6 +47,7 @@ Forge will call `ext.Start(ctx)` on application start and `ext.Stop(ctx)` on shu | `WithDisableMigrate()` | Skip auto-migration at startup | `false` | | `WithGroveDatabase(name)` | Resolve a grove.DB from DI by name | `""` | | `WithGroveKV(name)` | Resolve a grove KV store from DI by name | `""` | +| `WithKVKeyPrefix(prefix)` | Namespace the Redis store's keys per tenant (grove KV path only) | `""` | ## Auto-migration @@ -114,6 +115,19 @@ ext := extension.New( ) ``` +When several dispatch instances share one Redis, give each its own key +prefix. Every key, stream and pub/sub channel the store touches goes under +it (`ws_acme:dispatch:job:`), so instances never see each other's +queues, cron locks or leadership. Leave it empty for a single instance and +the keys stay `dispatch:*` as before: + +```go +ext := extension.New( + extension.WithGroveKV("dispatch-kv"), + extension.WithKVKeyPrefix("ws_acme:"), +) +``` + ### Store resolution order The extension resolves its store in this order: @@ -134,6 +148,7 @@ extensions: base_path: /api/dispatch grove_database: jobs grove_kv: dispatch-kv + key_prefix: "ws_acme:" concurrency: 20 queues: - default @@ -161,6 +176,7 @@ dispatch: | `BasePath` | `base_path` | `string` | `"/api/dispatch"` | URL prefix for all dispatch HTTP routes | | `GroveDatabase` | `grove_database` | `string` | `""` | Name of the grove.DB to resolve from DI; empty uses the default DB | | `GroveKV` | `grove_kv` | `string` | `""` | Name of the grove KV store to resolve from DI; empty uses the default KV | +| `KeyPrefix` | `key_prefix` | `string` | `""` | Tenant prefix for every Redis key the grove KV store writes; empty keeps `dispatch:*` | | `Concurrency` | `concurrency` | `int` | `10` | Max concurrent jobs | | `Queues` | `queues` | `[]string` | `["default"]` | Queues to poll | | `DisableRoutes` | `disable_routes` | `bool` | `false` | Skip HTTP route registration | diff --git a/docs/content/docs/stores/redis.mdx b/docs/content/docs/stores/redis.mdx index 3080745..f4b3d6b 100644 --- a/docs/content/docs/stores/redis.mdx +++ b/docs/content/docs/stores/redis.mdx @@ -27,6 +27,22 @@ if err := s.Ping(ctx); err != nil { d, err := dispatch.New(dispatch.WithStore(s)) ``` +## Sharing one Redis between instances + +All keys sit under `dispatch:`. To run more than one dispatch instance +against the same Redis, give each a key prefix and the store puts every +key, event stream and wake channel under it: + +```go +s := redis.New(kvStore, redis.WithKeyPrefix("ws_acme:")) +// jobs land at ws_acme:dispatch:job:, the queue at ws_acme:dispatch:queue: +``` + +An instance with a prefix never dequeues another instance's jobs, never +sees its cron names as duplicates, and takes leadership independently. An +empty prefix keeps the historical unprefixed keys, so existing deployments +carry on unchanged. + ## Internals | Aspect | Detail | diff --git a/extension/config.go b/extension/config.go index 4cb17b2..b6ad13b 100644 --- a/extension/config.go +++ b/extension/config.go @@ -50,6 +50,12 @@ type Config struct { // (unnamed) kv.Store is used. GroveKV string `json:"grove_kv" mapstructure:"grove_kv" yaml:"grove_kv"` + // KeyPrefix namespaces every key the Redis-backed store writes, so + // several dispatch instances can share one Redis without seeing each + // other's jobs, cron locks or leadership. Only the grove KV path reads + // it. Empty keeps the historical "dispatch:" keys. + KeyPrefix string `json:"key_prefix" mapstructure:"key_prefix" yaml:"key_prefix"` + // Artifacts configures the artifact plane. Artifacts ArtifactConfig `json:"artifacts" mapstructure:"artifacts" yaml:"artifacts"` diff --git a/extension/extension.go b/extension/extension.go index 0c3b0ea..8f04b95 100644 --- a/extension/extension.go +++ b/extension/extension.go @@ -165,7 +165,7 @@ func (e *Extension) init(fapp forge.App) error { if err != nil { return fmt.Errorf("dispatch: %w", err) } - e.dispatchOpts = append(e.dispatchOpts, dispatch.WithStore(redisstore.New(kvStore))) + e.dispatchOpts = append(e.dispatchOpts, dispatch.WithStore(e.buildStoreFromGroveKV(kvStore))) } else if db, err := vessel.Inject[*grove.DB](fapp.Container()); err == nil { // Auto-discover default grove.DB from container (matches authsome/cortex pattern). s, err := e.buildStoreFromGroveDB(db) @@ -498,6 +498,7 @@ func (e *Extension) loadConfiguration() error { forge.F("base_path", e.config.BasePath), forge.F("grove_database", e.config.GroveDatabase), forge.F("grove_kv", e.config.GroveKV), + forge.F("key_prefix", e.config.KeyPrefix), ) return nil @@ -624,6 +625,9 @@ func (e *Extension) mergeConfigurations(yamlConfig, programmaticConfig Config) C if yamlConfig.GroveKV == "" && programmaticConfig.GroveKV != "" { yamlConfig.GroveKV = programmaticConfig.GroveKV } + if yamlConfig.KeyPrefix == "" && programmaticConfig.KeyPrefix != "" { + yamlConfig.KeyPrefix = programmaticConfig.KeyPrefix + } if yamlConfig.DWPBasePath == "" && programmaticConfig.DWPBasePath != "" { yamlConfig.DWPBasePath = programmaticConfig.DWPBasePath } @@ -665,6 +669,12 @@ func (e *Extension) buildStoreFromGroveDB(db *grove.DB) (dispatch.Storer, error) } } +// buildStoreFromGroveKV wraps a grove KV store in the Redis backend, +// namespaced by the configured key prefix. +func (e *Extension) buildStoreFromGroveKV(kvStore *kv.Store) dispatch.Storer { + return redisstore.New(kvStore, redisstore.WithKeyPrefix(e.config.KeyPrefix)) +} + // resolveGroveKV resolves a *kv.Store from the DI container. // If GroveKV is set, it looks up the named store; otherwise it uses the default. func (e *Extension) resolveGroveKV(fapp forge.App) (*kv.Store, error) { diff --git a/extension/kvprefix_internal_test.go b/extension/kvprefix_internal_test.go new file mode 100644 index 0000000..aa06609 --- /dev/null +++ b/extension/kvprefix_internal_test.go @@ -0,0 +1,66 @@ +package extension + +import ( + "testing" + + "github.com/xraph/grove/kv" + "github.com/xraph/grove/kv/drivers/redisdriver" + + redisstore "github.com/xraph/dispatch/store/redis" +) + +// unopenedKV returns a Redis-driver KV store that never dials. Building the +// dispatch store needs only the driver type, so the tests below can pin the +// wiring without a Redis on the box. +func unopenedKV(t *testing.T) *kv.Store { + t.Helper() + s, err := kv.Open(redisdriver.New()) + if err != nil { + t.Fatalf("open kv: %v", err) + } + return s +} + +func TestBuildStoreFromGroveKV_appliesKeyPrefix(t *testing.T) { + tests := []struct { + name string + prefix string + }{ + {name: "no prefix keeps the historical keys", prefix: ""}, + {name: "tenant prefix reaches the redis store", prefix: "ws_acme:"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + e := New(WithKVKeyPrefix(tt.prefix)) + s, ok := e.buildStoreFromGroveKV(unopenedKV(t)).(*redisstore.Store) + if !ok { + t.Fatal("grove KV must build a redis store") + } + if got := s.KeyPrefix(); got != tt.prefix { + t.Fatalf("store key prefix = %q, want %q", got, tt.prefix) + } + }) + } +} + +func TestMergeConfigurations_keyPrefixYAMLWins(t *testing.T) { + e := New() + tests := []struct { + name string + yaml string + programmatic string + want string + }{ + {name: "yaml set, programmatic empty", yaml: "ws_yaml:", programmatic: "", want: "ws_yaml:"}, + {name: "yaml empty, programmatic set", yaml: "", programmatic: "ws_go:", want: "ws_go:"}, + {name: "both set, yaml wins", yaml: "ws_yaml:", programmatic: "ws_go:", want: "ws_yaml:"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := e.mergeConfigurations(Config{KeyPrefix: tt.yaml}, Config{KeyPrefix: tt.programmatic}) + if got.KeyPrefix != tt.want { + t.Fatalf("merged KeyPrefix = %q, want %q", got.KeyPrefix, tt.want) + } + }) + } +} diff --git a/extension/options.go b/extension/options.go index 02d8cb4..1f66681 100644 --- a/extension/options.go +++ b/extension/options.go @@ -205,6 +205,15 @@ func WithGroveKV(name string) ExtOption { } } +// WithKVKeyPrefix namespaces the Redis-backed store's keys (see +// Config.KeyPrefix). Pair it with WithGroveKV; the grove database path +// ignores it. +func WithKVKeyPrefix(prefix string) ExtOption { + return func(e *Extension) { + e.config.KeyPrefix = prefix + } +} + // WithDWP enables the Dispatch Wire Protocol (DWP) for real-time // client communication over WebSocket, SSE, and HTTP RPC. // Options configure authentication, codec, and server behaviour. diff --git a/store/redis/artifact.go b/store/redis/artifact.go index 6441917..a5df9ee 100644 --- a/store/redis/artifact.go +++ b/store/redis/artifact.go @@ -130,7 +130,7 @@ func linkField(name string, attempt int) string { // two concurrent creates at the same coordinates resolve to one winner // and one ErrExists. func (s *Store) CreateArtifact(ctx context.Context, a *artifact.Artifact, link *artifact.Link) error { - guard := artifactKeyGuard(a.Backend, a.Bucket, a.Key) + guard := s.keys.artifactGuard(a.Backend, a.Bucket, a.Key) ok, err := s.rdb.SetNX(ctx, guard, a.ID.String(), 0).Result() if err != nil { @@ -141,14 +141,14 @@ func (s *Store) CreateArtifact(ctx context.Context, a *artifact.Artifact, link * return artifact.ErrExists } - if err := s.setEntity(ctx, artifactKey(a.ID.String()), toArtifactEntity(a)); err != nil { + if err := s.setEntity(ctx, s.keys.artifact(a.ID.String()), toArtifactEntity(a)); err != nil { // Release the guard so the coordinates are not permanently burned. s.rdb.Del(ctx, guard) return fmt.Errorf("dispatch/redis: create artifact: %w", err) } - if err := s.rdb.SAdd(ctx, artifactIDsKey, a.ID.String()).Err(); err != nil { + if err := s.rdb.SAdd(ctx, s.keys.artifactIDs(), a.ID.String()).Err(); err != nil { return fmt.Errorf("dispatch/redis: index artifact: %w", err) } @@ -157,7 +157,7 @@ func (s *Store) CreateArtifact(ctx context.Context, a *artifact.Artifact, link * // no durable artifact is ever a member. if a.Lifecycle == artifact.Ephemeral { score := float64(a.CreatedAt.UnixNano()) - if err := s.rdb.ZAdd(ctx, artifactEphemeralKey, goredis.Z{Score: score, Member: a.ID.String()}).Err(); err != nil { + if err := s.rdb.ZAdd(ctx, s.keys.artifactEphemeral(), goredis.Z{Score: score, Member: a.ID.String()}).Err(); err != nil { return fmt.Errorf("dispatch/redis: index ephemeral artifact: %w", err) } } @@ -187,7 +187,7 @@ func (s *Store) GetArtifact(ctx context.Context, artifactID id.ArtifactID) (*art func (s *Store) loadArtifact(ctx context.Context, artifactID string) (*artifact.Artifact, error) { var e artifactEntity - if err := s.getEntity(ctx, artifactKey(artifactID), &e); err != nil { + if err := s.getEntity(ctx, s.keys.artifact(artifactID), &e); err != nil { if isNotFound(err) { return nil, artifact.ErrNotFound } @@ -200,7 +200,7 @@ func (s *Store) loadArtifact(ctx context.Context, artifactID string) (*artifact. // FindArtifactByKey retrieves a live artifact by its storage coordinates. func (s *Store) FindArtifactByKey(ctx context.Context, backend, bucket, key string) (*artifact.Artifact, error) { - got, err := s.rdb.Get(ctx, artifactKeyGuard(backend, bucket, key)).Result() + got, err := s.rdb.Get(ctx, s.keys.artifactGuard(backend, bucket, key)).Result() if err != nil { return nil, artifact.ErrNotFound } @@ -221,7 +221,7 @@ func (s *Store) UpdateArtifact(ctx context.Context, a *artifact.Artifact) error existing.ContentType = a.ContentType existing.ExpiresAt = a.ExpiresAt - if err := s.setEntity(ctx, artifactKey(a.ID.String()), toArtifactEntity(existing)); err != nil { + if err := s.setEntity(ctx, s.keys.artifact(a.ID.String()), toArtifactEntity(existing)); err != nil { return fmt.Errorf("dispatch/redis: update artifact: %w", err) } @@ -230,7 +230,7 @@ func (s *Store) UpdateArtifact(ctx context.Context, a *artifact.Artifact) error // ListArtifacts returns artifacts matching the given options, newest first. func (s *Store) ListArtifacts(ctx context.Context, opts artifact.ListOpts) ([]*artifact.Artifact, error) { - ids, err := s.rdb.SMembers(ctx, artifactIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.artifactIDs()).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: list artifact ids: %w", err) } @@ -300,12 +300,12 @@ func (s *Store) LinkArtifact(ctx context.Context, link *artifact.Link) error { owner := artifact.OwnerRef{Kind: link.OwnerKind, ID: link.OwnerID} field := linkField(link.Name, link.Attempt) - if err := s.rdb.HSet(ctx, ownerLinksKey(string(owner.Kind), owner.ID), field, raw).Err(); err != nil { + if err := s.rdb.HSet(ctx, s.keys.ownerLinks(string(owner.Kind), owner.ID), field, raw).Err(); err != nil { return fmt.Errorf("dispatch/redis: link artifact: %w", err) } member := string(link.OwnerKind) + "\x00" + link.OwnerID + "\x00" + field - if err := s.rdb.SAdd(ctx, artifactLinksKey(link.ArtifactID.String()), member).Err(); err != nil { + if err := s.rdb.SAdd(ctx, s.keys.artifactLinks(link.ArtifactID.String()), member).Err(); err != nil { return fmt.Errorf("dispatch/redis: index artifact link: %w", err) } @@ -314,7 +314,7 @@ func (s *Store) LinkArtifact(ctx context.Context, link *artifact.Link) error { // ListLinks returns every link belonging to the given owner. func (s *Store) ListLinks(ctx context.Context, owner artifact.OwnerRef) ([]*artifact.Link, error) { - vals, err := s.rdb.HGetAll(ctx, ownerLinksKey(string(owner.Kind), owner.ID)).Result() + vals, err := s.rdb.HGetAll(ctx, s.keys.ownerLinks(string(owner.Kind), owner.ID)).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: list links: %w", err) } @@ -449,7 +449,7 @@ func (s *Store) SweepEphemeral( limit = defaultSweepLimit } - ids, err := s.rdb.ZRange(ctx, artifactEphemeralKey, 0, -1).Result() + ids, err := s.rdb.ZRange(ctx, s.keys.artifactEphemeral(), 0, -1).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: sweep ephemeral: %w", err) } @@ -515,7 +515,7 @@ func (s *Store) SweepEphemeral( // linksForArtifact resolves every link pointing at an artifact. func (s *Store) linksForArtifact(ctx context.Context, artifactID string) ([]*artifact.Link, error) { - members, err := s.rdb.SMembers(ctx, artifactLinksKey(artifactID)).Result() + members, err := s.rdb.SMembers(ctx, s.keys.artifactLinks(artifactID)).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: list artifact links: %w", err) } @@ -530,7 +530,7 @@ func (s *Store) linksForArtifact(ctx context.Context, artifactID string) ([]*art owner := artifact.OwnerRef{Kind: artifact.OwnerKind(kind), ID: ownerID} - raw, herr := s.rdb.HGet(ctx, ownerLinksKey(string(owner.Kind), owner.ID), field).Result() + raw, herr := s.rdb.HGet(ctx, s.keys.ownerLinks(string(owner.Kind), owner.ID), field).Result() if herr != nil { continue } @@ -604,7 +604,7 @@ func (s *Store) ownerTerminalAt(ctx context.Context, l *artifact.Link) (time.Tim switch l.OwnerKind { case artifact.OwnerJob: var e jobEntity - if err := s.getEntity(ctx, jobKey(l.OwnerID), &e); err != nil { + if err := s.getEntity(ctx, s.keys.job(l.OwnerID), &e); err != nil { if isNotFound(err) { return l.CreatedAt, true, nil } @@ -624,7 +624,7 @@ func (s *Store) ownerTerminalAt(ctx context.Context, l *artifact.Link) (time.Tim case artifact.OwnerRun, artifact.OwnerStep: var e runEntity - if err := s.getEntity(ctx, runKey(l.OwnerID), &e); err != nil { + if err := s.getEntity(ctx, s.keys.run(l.OwnerID), &e); err != nil { if isNotFound(err) { return l.CreatedAt, true, nil } @@ -668,7 +668,7 @@ func (s *Store) SweepOrphans( // The ephemeral index is scored by creation time, so the cutoff is a // range query rather than a scan. - ids, err := s.rdb.ZRangeByScore(ctx, artifactEphemeralKey, &goredis.ZRangeBy{ + ids, err := s.rdb.ZRangeByScore(ctx, s.keys.artifactEphemeral(), &goredis.ZRangeBy{ Min: "-inf", Max: strconv.FormatInt(cutoff.UnixNano(), 10), }).Result() @@ -700,7 +700,7 @@ func (s *Store) SweepOrphans( continue } - n, cerr := s.rdb.SCard(ctx, artifactLinksKey(a.ID.String())).Result() + n, cerr := s.rdb.SCard(ctx, s.keys.artifactLinks(a.ID.String())).Result() if cerr != nil { return nil, fmt.Errorf("dispatch/redis: count artifact links: %w", cerr) } @@ -737,20 +737,20 @@ func (s *Store) markDeleted( deleted := at clone.DeletedAt = &deleted - if err := s.setEntity(ctx, artifactKey(clone.ID.String()), toArtifactEntity(clone)); err != nil { + if err := s.setEntity(ctx, s.keys.artifact(clone.ID.String()), toArtifactEntity(clone)); err != nil { return nil, fmt.Errorf("dispatch/redis: mark artifact deleted: %w", err) } // Release the live-key guard so the coordinates become reusable, // and index the deletion time for the purge pass. - s.rdb.Del(ctx, artifactKeyGuard(clone.Backend, clone.Bucket, clone.Key)) + s.rdb.Del(ctx, s.keys.artifactGuard(clone.Backend, clone.Bucket, clone.Key)) - if err := s.rdb.ZAdd(ctx, artifactDeletedKey, + if err := s.rdb.ZAdd(ctx, s.keys.artifactDeleted(), goredis.Z{Score: float64(at.UnixNano()), Member: clone.ID.String()}).Err(); err != nil { return nil, fmt.Errorf("dispatch/redis: index deleted artifact: %w", err) } - if err := s.rdb.ZRem(ctx, artifactEphemeralKey, clone.ID.String()).Err(); err != nil { + if err := s.rdb.ZRem(ctx, s.keys.artifactEphemeral(), clone.ID.String()).Err(); err != nil { return nil, fmt.Errorf("dispatch/redis: deindex ephemeral artifact: %w", err) } @@ -772,7 +772,7 @@ func (s *Store) ListPurgeable( cutoff := time.Now().UTC().Add(-grace) - ids, err := s.rdb.ZRangeByScore(ctx, artifactDeletedKey, &goredis.ZRangeBy{ + ids, err := s.rdb.ZRangeByScore(ctx, s.keys.artifactDeleted(), &goredis.ZRangeBy{ Min: "-inf", Max: strconv.FormatInt(cutoff.UnixNano(), 10), }).Result() @@ -817,22 +817,22 @@ func (s *Store) PurgeArtifact(ctx context.Context, artifactID id.ArtifactID) err for _, l := range links { owner := artifact.OwnerRef{Kind: l.OwnerKind, ID: l.OwnerID} - if herr := s.rdb.HDel(ctx, ownerLinksKey(string(owner.Kind), owner.ID), linkField(l.Name, l.Attempt)).Err(); herr != nil { + if herr := s.rdb.HDel(ctx, s.keys.ownerLinks(string(owner.Kind), owner.ID), linkField(l.Name, l.Attempt)).Err(); herr != nil { return fmt.Errorf("dispatch/redis: purge artifact link: %w", herr) } } a, err := s.loadArtifact(ctx, key) if err == nil { - s.rdb.Del(ctx, artifactKeyGuard(a.Backend, a.Bucket, a.Key)) + s.rdb.Del(ctx, s.keys.artifactGuard(a.Backend, a.Bucket, a.Key)) } pipe := s.rdb.TxPipeline() - pipe.Del(ctx, artifactKey(key)) - pipe.Del(ctx, artifactLinksKey(key)) - pipe.SRem(ctx, artifactIDsKey, key) - pipe.ZRem(ctx, artifactEphemeralKey, key) - pipe.ZRem(ctx, artifactDeletedKey, key) + pipe.Del(ctx, s.keys.artifact(key)) + pipe.Del(ctx, s.keys.artifactLinks(key)) + pipe.SRem(ctx, s.keys.artifactIDs(), key) + pipe.ZRem(ctx, s.keys.artifactEphemeral(), key) + pipe.ZRem(ctx, s.keys.artifactDeleted(), key) if _, err := pipe.Exec(ctx); err != nil { return fmt.Errorf("dispatch/redis: purge artifact: %w", err) diff --git a/store/redis/cluster.go b/store/redis/cluster.go index c638299..45f0668 100644 --- a/store/redis/cluster.go +++ b/store/redis/cluster.go @@ -63,14 +63,14 @@ func fromWorkerEntity(e *workerEntity) (*cluster.Worker, error) { // RegisterWorker adds a new worker to the cluster registry. func (s *Store) RegisterWorker(ctx context.Context, w *cluster.Worker) error { wID := w.ID.String() - key := workerKey(wID) + key := s.keys.worker(wID) e := toWorkerEntity(w) if err := s.setEntity(ctx, key, e); err != nil { return fmt.Errorf("dispatch/redis: register worker set: %w", err) } - if err := s.rdb.SAdd(ctx, workerIDsKey, wID).Err(); err != nil { + if err := s.rdb.SAdd(ctx, s.keys.workerIDs(), wID).Err(); err != nil { return fmt.Errorf("dispatch/redis: register worker index: %w", err) } return nil @@ -79,7 +79,7 @@ func (s *Store) RegisterWorker(ctx context.Context, w *cluster.Worker) error { // DeregisterWorker removes a worker from the cluster registry. func (s *Store) DeregisterWorker(ctx context.Context, workerID id.WorkerID) error { wID := workerID.String() - key := workerKey(wID) + key := s.keys.worker(wID) exists, err := s.entityExists(ctx, key) if err != nil { @@ -91,7 +91,7 @@ func (s *Store) DeregisterWorker(ctx context.Context, workerID id.WorkerID) erro pipe := s.rdb.TxPipeline() pipe.Del(ctx, key) - pipe.SRem(ctx, workerIDsKey, wID) + pipe.SRem(ctx, s.keys.workerIDs(), wID) _, err = pipe.Exec(ctx) if err != nil { return fmt.Errorf("dispatch/redis: deregister worker: %w", err) @@ -101,7 +101,7 @@ func (s *Store) DeregisterWorker(ctx context.Context, workerID id.WorkerID) erro // HeartbeatWorker updates the last-seen timestamp for a worker. func (s *Store) HeartbeatWorker(ctx context.Context, workerID id.WorkerID) error { - key := workerKey(workerID.String()) + key := s.keys.worker(workerID.String()) var e workerEntity if err := s.getEntity(ctx, key, &e); err != nil { @@ -117,7 +117,7 @@ func (s *Store) HeartbeatWorker(ctx context.Context, workerID id.WorkerID) error // ListWorkers returns all registered workers. func (s *Store) ListWorkers(ctx context.Context) ([]*cluster.Worker, error) { - ids, err := s.rdb.SMembers(ctx, workerIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.workerIDs()).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: list workers: %w", err) } @@ -125,7 +125,7 @@ func (s *Store) ListWorkers(ctx context.Context) ([]*cluster.Worker, error) { workers := make([]*cluster.Worker, 0, len(ids)) for _, wID := range ids { var e workerEntity - if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr != nil { continue } w, convErr := fromWorkerEntity(&e) @@ -142,7 +142,7 @@ func (s *Store) ListWorkers(ctx context.Context) ([]*cluster.Worker, error) { func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration) (int64, error) { cutoff := now().Add(-threshold) - ids, err := s.rdb.SMembers(ctx, workerIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.workerIDs()).Result() if err != nil { return 0, fmt.Errorf("dispatch/redis: delete stale smembers: %w", err) } @@ -150,11 +150,11 @@ func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration) var deleted int64 for _, wID := range ids { var e workerEntity - if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr != nil { // Orphaned set member without a backing entity — treat as // stale and remove from the index. if isNotFound(getErr) { - if remErr := s.rdb.SRem(ctx, workerIDsKey, wID).Err(); remErr == nil { + if remErr := s.rdb.SRem(ctx, s.keys.workerIDs(), wID).Err(); remErr == nil { deleted++ } } @@ -162,8 +162,8 @@ func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration) } if e.LastSeen.Before(cutoff) { pipe := s.rdb.TxPipeline() - pipe.Del(ctx, workerKey(wID)) - pipe.SRem(ctx, workerIDsKey, wID) + pipe.Del(ctx, s.keys.worker(wID)) + pipe.SRem(ctx, s.keys.workerIDs(), wID) if _, execErr := pipe.Exec(ctx); execErr == nil { deleted++ } @@ -176,7 +176,7 @@ func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration) func (s *Store) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([]*cluster.Worker, error) { cutoff := now().Add(-threshold) - ids, err := s.rdb.SMembers(ctx, workerIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.workerIDs()).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: reap smembers: %w", err) } @@ -184,7 +184,7 @@ func (s *Store) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([ var dead []*cluster.Worker for _, wID := range ids { var e workerEntity - if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr != nil { continue } if e.LastSeen.Before(cutoff) { @@ -201,7 +201,7 @@ func (s *Store) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([ // AcquireLeadership attempts to become the cluster leader. func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error) { wID := workerID.String() - wKey := workerKey(wID) + wKey := s.keys.worker(wID) // Check worker exists. exists, err := s.entityExists(ctx, wKey) @@ -213,7 +213,7 @@ func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl } // Try SET NX with TTL (atomic acquire). - ok, err := s.rdb.SetNX(ctx, leaderKey, wID, ttl).Result() + ok, err := s.rdb.SetNX(ctx, s.keys.leader(), wID, ttl).Result() if err != nil { return false, fmt.Errorf("dispatch/redis: acquire leadership setnx: %w", err) } @@ -230,13 +230,13 @@ func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl } // Check if we already hold it. - current, err := s.rdb.Get(ctx, leaderKey).Result() + current, err := s.rdb.Get(ctx, s.keys.leader()).Result() if err != nil && !isRedisNil(err) { return false, fmt.Errorf("dispatch/redis: acquire leadership get: %w", err) } if current == wID { // Re-acquire: extend TTL. - _ = s.rdb.Expire(ctx, leaderKey, ttl).Err() //nolint:errcheck // best-effort + _ = s.rdb.Expire(ctx, s.keys.leader(), ttl).Err() //nolint:errcheck // best-effort until := now().Add(ttl) var e workerEntity if getErr := s.getEntity(ctx, wKey, &e); getErr == nil { @@ -254,7 +254,7 @@ func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl func (s *Store) RenewLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error) { wID := workerID.String() - current, err := s.rdb.Get(ctx, leaderKey).Result() + current, err := s.rdb.Get(ctx, s.keys.leader()).Result() if err != nil { if isRedisNil(err) { return false, nil // no leader @@ -265,19 +265,19 @@ func (s *Store) RenewLeadership(ctx context.Context, workerID id.WorkerID, ttl t return false, nil // not the leader } - _ = s.rdb.Expire(ctx, leaderKey, ttl).Err() //nolint:errcheck // best-effort + _ = s.rdb.Expire(ctx, s.keys.leader(), ttl).Err() //nolint:errcheck // best-effort until := now().Add(ttl) var e workerEntity - if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr == nil { + if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr == nil { e.LeaderUntil = &until - _ = s.setEntity(ctx, workerKey(wID), &e) //nolint:errcheck // best-effort update + _ = s.setEntity(ctx, s.keys.worker(wID), &e) //nolint:errcheck // best-effort update } return true, nil } // GetLeader returns the current cluster leader, or nil if there is no leader. func (s *Store) GetLeader(ctx context.Context) (*cluster.Worker, error) { - wID, err := s.rdb.Get(ctx, leaderKey).Result() + wID, err := s.rdb.Get(ctx, s.keys.leader()).Result() if err != nil { if isRedisNil(err) { return nil, nil // no leader @@ -286,7 +286,7 @@ func (s *Store) GetLeader(ctx context.Context) (*cluster.Worker, error) { } var e workerEntity - if getErr := s.getEntity(ctx, workerKey(wID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.worker(wID), &e); getErr != nil { return nil, nil // leader key exists but worker gone } return fromWorkerEntity(&e) diff --git a/store/redis/cron.go b/store/redis/cron.go index 52edf4b..1f2c388 100644 --- a/store/redis/cron.go +++ b/store/redis/cron.go @@ -80,10 +80,10 @@ func fromCronEntity(e *cronEntity) (*cron.Entry, error) { // RegisterCron persists a new cron entry. func (s *Store) RegisterCron(ctx context.Context, entry *cron.Entry) error { eID := entry.ID.String() - key := cronKey(eID) + key := s.keys.cron(eID) // Check for duplicate name. - existing, err := s.rdb.HGet(ctx, cronNamesKey, entry.Name).Result() + existing, err := s.rdb.HGet(ctx, s.keys.cronNames(), entry.Name).Result() if err != nil && !isRedisNil(err) { return fmt.Errorf("dispatch/redis: register cron check name: %w", err) } @@ -97,8 +97,8 @@ func (s *Store) RegisterCron(ctx context.Context, entry *cron.Entry) error { } pipe := s.rdb.TxPipeline() - pipe.SAdd(ctx, cronIDsKey, eID) - pipe.HSet(ctx, cronNamesKey, entry.Name, eID) + pipe.SAdd(ctx, s.keys.cronIDs(), eID) + pipe.HSet(ctx, s.keys.cronNames(), entry.Name, eID) _, err = pipe.Exec(ctx) if err != nil { return fmt.Errorf("dispatch/redis: register cron indexes: %w", err) @@ -109,7 +109,7 @@ func (s *Store) RegisterCron(ctx context.Context, entry *cron.Entry) error { // GetCron retrieves a cron entry by ID. func (s *Store) GetCron(ctx context.Context, entryID id.CronID) (*cron.Entry, error) { var e cronEntity - if err := s.getEntity(ctx, cronKey(entryID.String()), &e); err != nil { + if err := s.getEntity(ctx, s.keys.cron(entryID.String()), &e); err != nil { if isNotFound(err) { return nil, dispatch.ErrCronNotFound } @@ -120,7 +120,7 @@ func (s *Store) GetCron(ctx context.Context, entryID id.CronID) (*cron.Entry, er // ListCrons returns all cron entries. func (s *Store) ListCrons(ctx context.Context) ([]*cron.Entry, error) { - ids, err := s.rdb.SMembers(ctx, cronIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.cronIDs()).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: list crons: %w", err) } @@ -128,7 +128,7 @@ func (s *Store) ListCrons(ctx context.Context) ([]*cron.Entry, error) { entries := make([]*cron.Entry, 0, len(ids)) for _, eID := range ids { var e cronEntity - if getErr := s.getEntity(ctx, cronKey(eID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.cron(eID), &e); getErr != nil { continue } entry, convErr := fromCronEntity(&e) @@ -143,7 +143,7 @@ func (s *Store) ListCrons(ctx context.Context) ([]*cron.Entry, error) { // AcquireCronLock attempts to acquire a distributed lock for a cron entry. func (s *Store) AcquireCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID, ttl time.Duration) (bool, error) { eID := entryID.String() - key := cronKey(eID) + key := s.keys.cron(eID) wID := workerID.String() t := now() until := t.Add(ttl) @@ -177,7 +177,7 @@ func (s *Store) AcquireCronLock(ctx context.Context, entryID id.CronID, workerID // ReleaseCronLock releases the distributed lock for a cron entry. func (s *Store) ReleaseCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID) error { - key := cronKey(entryID.String()) + key := s.keys.cron(entryID.String()) wID := workerID.String() var e cronEntity @@ -200,7 +200,7 @@ func (s *Store) ReleaseCronLock(ctx context.Context, entryID id.CronID, workerID // UpdateCronLastRun records when a cron entry last fired. func (s *Store) UpdateCronLastRun(ctx context.Context, entryID id.CronID, at time.Time) error { - key := cronKey(entryID.String()) + key := s.keys.cron(entryID.String()) var e cronEntity if err := s.getEntity(ctx, key, &e); err != nil { if isNotFound(err) { @@ -216,7 +216,7 @@ func (s *Store) UpdateCronLastRun(ctx context.Context, entryID id.CronID, at tim // UpdateCronEntry updates a cron entry. func (s *Store) UpdateCronEntry(ctx context.Context, entry *cron.Entry) error { - key := cronKey(entry.ID.String()) + key := s.keys.cron(entry.ID.String()) exists, err := s.entityExists(ctx, key) if err != nil { return fmt.Errorf("dispatch/redis: update cron exists: %w", err) @@ -233,7 +233,7 @@ func (s *Store) UpdateCronEntry(ctx context.Context, entry *cron.Entry) error { // DeleteCron removes a cron entry by ID. func (s *Store) DeleteCron(ctx context.Context, entryID id.CronID) error { eID := entryID.String() - key := cronKey(eID) + key := s.keys.cron(eID) // Get name for name index cleanup. var e cronEntity @@ -246,9 +246,9 @@ func (s *Store) DeleteCron(ctx context.Context, entryID id.CronID) error { pipe := s.rdb.TxPipeline() pipe.Del(ctx, key) - pipe.SRem(ctx, cronIDsKey, eID) + pipe.SRem(ctx, s.keys.cronIDs(), eID) if e.Name != "" { - pipe.HDel(ctx, cronNamesKey, e.Name) + pipe.HDel(ctx, s.keys.cronNames(), e.Name) } _, err := pipe.Exec(ctx) if err != nil { diff --git a/store/redis/dequeue.go b/store/redis/dequeue.go index 59277fa..116fcdb 100644 --- a/store/redis/dequeue.go +++ b/store/redis/dequeue.go @@ -233,7 +233,7 @@ func (s *Store) scanQueue( q string, t time.Time, ) ([]dequeueCandidate, error) { - key := queueKey(q) + key := s.keys.queue(q) full := !opts.IsUnbounded() || len(opts.PreferredHashes()) > 0 var ( @@ -353,7 +353,7 @@ func (s *Store) readJobEntities(ctx context.Context, ids []string) ([]*jobEntity cmds := make([]*goredis.StringCmd, len(ids)) for i, jID := range ids { - cmds[i] = pipe.Get(ctx, jobKey(jID)) + cmds[i] = pipe.Get(ctx, s.keys.job(jID)) } // A missing key makes Exec report goredis.Nil for the batch as a @@ -429,7 +429,7 @@ func (s *Store) claimCandidates( rems := make([]*goredis.IntCmd, len(candidates)) for i, c := range candidates { - rems[i] = pipe.ZRem(ctx, queueKey(c.queue), c.id) + rems[i] = pipe.ZRem(ctx, s.keys.queue(c.queue), c.id) } if _, err := pipe.Exec(ctx); err != nil { @@ -445,7 +445,7 @@ func (s *Store) claimCandidates( continue // another worker removed it first } - key := jobKey(c.id) + key := s.keys.job(c.id) var e jobEntity if getErr := s.getEntity(ctx, key, &e); getErr != nil { diff --git a/store/redis/dlq.go b/store/redis/dlq.go index 165f573..d79863b 100644 --- a/store/redis/dlq.go +++ b/store/redis/dlq.go @@ -108,14 +108,14 @@ func fromDLQEntity(e *dlqEntity) (*dlq.Entry, error) { // PushDLQ adds a failed job entry to the dead letter queue. func (s *Store) PushDLQ(ctx context.Context, entry *dlq.Entry) error { eID := entry.ID.String() - key := dlqKey(eID) + key := s.keys.dlq(eID) e := toDLQEntity(entry) if err := s.setEntity(ctx, key, e); err != nil { return fmt.Errorf("dispatch/redis: push dlq set: %w", err) } - if err := s.rdb.SAdd(ctx, dlqIDsKey, eID).Err(); err != nil { + if err := s.rdb.SAdd(ctx, s.keys.dlqIDs(), eID).Err(); err != nil { return fmt.Errorf("dispatch/redis: push dlq index: %w", err) } return nil @@ -123,7 +123,7 @@ func (s *Store) PushDLQ(ctx context.Context, entry *dlq.Entry) error { // ListDLQ returns DLQ entries matching the given options. func (s *Store) ListDLQ(ctx context.Context, opts dlq.ListOpts) ([]*dlq.Entry, error) { - ids, err := s.rdb.SMembers(ctx, dlqIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.dlqIDs()).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: list dlq: %w", err) } @@ -131,7 +131,7 @@ func (s *Store) ListDLQ(ctx context.Context, opts dlq.ListOpts) ([]*dlq.Entry, e entries := make([]*dlq.Entry, 0, len(ids)) for _, eID := range ids { var e dlqEntity - if getErr := s.getEntity(ctx, dlqKey(eID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.dlq(eID), &e); getErr != nil { continue } if opts.Queue != "" && e.Queue != opts.Queue { @@ -150,7 +150,7 @@ func (s *Store) ListDLQ(ctx context.Context, opts dlq.ListOpts) ([]*dlq.Entry, e // GetDLQ retrieves a DLQ entry by ID. func (s *Store) GetDLQ(ctx context.Context, entryID id.DLQID) (*dlq.Entry, error) { var e dlqEntity - if err := s.getEntity(ctx, dlqKey(entryID.String()), &e); err != nil { + if err := s.getEntity(ctx, s.keys.dlq(entryID.String()), &e); err != nil { if isNotFound(err) { return nil, dispatch.ErrDLQNotFound } @@ -161,7 +161,7 @@ func (s *Store) GetDLQ(ctx context.Context, entryID id.DLQID) (*dlq.Entry, error // ReplayDLQ marks a DLQ entry as replayed. func (s *Store) ReplayDLQ(ctx context.Context, entryID id.DLQID) error { - key := dlqKey(entryID.String()) + key := s.keys.dlq(entryID.String()) var e dlqEntity if err := s.getEntity(ctx, key, &e); err != nil { if isNotFound(err) { @@ -177,14 +177,14 @@ func (s *Store) ReplayDLQ(ctx context.Context, entryID id.DLQID) error { // PurgeDLQ removes DLQ entries with FailedAt before the given time. func (s *Store) PurgeDLQ(ctx context.Context, before time.Time) (int64, error) { - ids, err := s.rdb.SMembers(ctx, dlqIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.dlqIDs()).Result() if err != nil { return 0, fmt.Errorf("dispatch/redis: purge dlq smembers: %w", err) } var purged int64 for _, eID := range ids { - key := dlqKey(eID) + key := s.keys.dlq(eID) var e dlqEntity if getErr := s.getEntity(ctx, key, &e); getErr != nil { continue @@ -193,7 +193,7 @@ func (s *Store) PurgeDLQ(ctx context.Context, before time.Time) (int64, error) { if e.FailedAt.Before(before) { pipe := s.rdb.TxPipeline() pipe.Del(ctx, key) - pipe.SRem(ctx, dlqIDsKey, eID) + pipe.SRem(ctx, s.keys.dlqIDs(), eID) if _, pErr := pipe.Exec(ctx); pErr != nil { return purged, fmt.Errorf("dispatch/redis: purge dlq del: %w", pErr) } @@ -205,7 +205,7 @@ func (s *Store) PurgeDLQ(ctx context.Context, before time.Time) (int64, error) { // CountDLQ returns the total number of entries in the dead letter queue. func (s *Store) CountDLQ(ctx context.Context) (int64, error) { - count, err := s.rdb.SCard(ctx, dlqIDsKey).Result() + count, err := s.rdb.SCard(ctx, s.keys.dlqIDs()).Result() if err != nil { return 0, fmt.Errorf("dispatch/redis: count dlq: %w", err) } diff --git a/store/redis/event.go b/store/redis/event.go index 8bfb14d..678b425 100644 --- a/store/redis/event.go +++ b/store/redis/event.go @@ -56,7 +56,7 @@ func fromEventEntity(e *eventEntity) (*event.Event, error) { // PublishEvent persists a new event and adds it to the name's stream. func (s *Store) PublishEvent(ctx context.Context, evt *event.Event) error { eID := evt.ID.String() - key := eventKey(eID) + key := s.keys.event(eID) e := toEventEntity(evt) if err := s.setEntity(ctx, key, e); err != nil { @@ -65,7 +65,7 @@ func (s *Store) PublishEvent(ctx context.Context, evt *event.Event) error { // Add to the named stream so subscribers get notified. if err := s.rdb.XAdd(ctx, &goredis.XAddArgs{ - Stream: eventStreamKey(evt.Name), + Stream: s.keys.eventStream(evt.Name), Values: map[string]interface{}{ "event_id": eID, }, @@ -78,7 +78,7 @@ func (s *Store) PublishEvent(ctx context.Context, evt *event.Event) error { // SubscribeEvent waits for an unacked event matching the given name. // Uses stream polling for efficient waiting. func (s *Store) SubscribeEvent(ctx context.Context, name string, timeout time.Duration) (*event.Event, error) { - stream := eventStreamKey(name) + stream := s.keys.eventStream(name) deadline := time.Now().Add(timeout) for { @@ -104,7 +104,7 @@ func (s *Store) SubscribeEvent(ctx context.Context, name string, timeout time.Du continue } - key := eventKey(eID) + key := s.keys.event(eID) var e eventEntity if getErr := s.getEntity(ctx, key, &e); getErr != nil { continue @@ -132,7 +132,7 @@ func (s *Store) SubscribeEvent(ctx context.Context, name string, timeout time.Du // AckEvent acknowledges an event, marking it as consumed. func (s *Store) AckEvent(ctx context.Context, eventID id.EventID) error { - key := eventKey(eventID.String()) + key := s.keys.event(eventID.String()) var e eventEntity if err := s.getEntity(ctx, key, &e); err != nil { diff --git a/store/redis/job.go b/store/redis/job.go index adaab4f..ed325d0 100644 --- a/store/redis/job.go +++ b/store/redis/job.go @@ -181,7 +181,7 @@ func fromJobEntity(e *jobEntity) (*job.Job, error) { // EnqueueJob stores the job as a JSON entity and adds it to the queue's Sorted Set. func (s *Store) EnqueueJob(ctx context.Context, j *job.Job) error { jID := j.ID.String() - key := jobKey(jID) + key := s.keys.job(jID) // Check for duplicate. exists, err := s.entityExists(ctx, key) @@ -201,11 +201,11 @@ func (s *Store) EnqueueJob(ctx context.Context, j *job.Job) error { } pipe := s.rdb.TxPipeline() - pipe.SAdd(ctx, jobIDsKey, jID) + pipe.SAdd(ctx, s.keys.jobIDs(), jID) // Add to queue sorted set: score = priority (negated for DESC) + time component. score := jobScore(j.Priority, j.RunAt) - pipe.ZAdd(ctx, queueKey(j.Queue), goredis.Z{Score: score, Member: jID}) + pipe.ZAdd(ctx, s.keys.queue(j.Queue), goredis.Z{Score: score, Member: jID}) _, err = pipe.Exec(ctx) if err != nil { @@ -221,7 +221,7 @@ func (s *Store) EnqueueJob(ctx context.Context, j *job.Job) error { // GetJob retrieves a job by ID. func (s *Store) GetJob(ctx context.Context, jobID id.JobID) (*job.Job, error) { var e jobEntity - if err := s.getEntity(ctx, jobKey(jobID.String()), &e); err != nil { + if err := s.getEntity(ctx, s.keys.job(jobID.String()), &e); err != nil { if isNotFound(err) { return nil, dispatch.ErrJobNotFound } @@ -260,7 +260,7 @@ func (s *Store) GetJob(ctx context.Context, jobID id.JobID) (*job.Job, error) { // safe for a job that was never claimed. func (s *Store) UpdateJob(ctx context.Context, j *job.Job) error { jID := j.ID.String() - key := jobKey(jID) + key := s.keys.job(jID) exists, err := s.entityExists(ctx, key) if err != nil { @@ -280,7 +280,7 @@ func (s *Store) UpdateJob(ctx context.Context, j *job.Job) error { // so j.Queue is the queue it was indexed under. If that ever changes, // the old queue keeps a member pointing at this job, and this function // has to read the stored entity to learn which queue to clear. - qk := queueKey(j.Queue) + qk := s.keys.queue(j.Queue) runnable := j.State == job.StatePending || j.State == job.StateRetrying if runnable { @@ -306,7 +306,7 @@ func (s *Store) UpdateJob(ctx context.Context, j *job.Job) error { // DeleteJob removes a job by ID. func (s *Store) DeleteJob(ctx context.Context, jobID id.JobID) error { jID := jobID.String() - key := jobKey(jID) + key := s.keys.job(jID) // Get queue name before deleting to remove from sorted set. var e jobEntity @@ -320,8 +320,8 @@ func (s *Store) DeleteJob(ctx context.Context, jobID id.JobID) error { // Delete entity via raw Redis DEL (KV store may not have Delete). pipe := s.rdb.TxPipeline() pipe.Del(ctx, key) - pipe.SRem(ctx, jobIDsKey, jID) - pipe.ZRem(ctx, queueKey(e.Queue), jID) + pipe.SRem(ctx, s.keys.jobIDs(), jID) + pipe.ZRem(ctx, s.keys.queue(e.Queue), jID) _, err := pipe.Exec(ctx) if err != nil { return fmt.Errorf("dispatch/redis: delete job: %w", err) @@ -331,7 +331,7 @@ func (s *Store) DeleteJob(ctx context.Context, jobID id.JobID) error { // ListJobsByState returns jobs matching the given state. func (s *Store) ListJobsByState(ctx context.Context, state job.State, opts job.ListOpts) ([]*job.Job, error) { - ids, err := s.rdb.SMembers(ctx, jobIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.jobIDs()).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: list jobs smembers: %w", err) } @@ -339,7 +339,7 @@ func (s *Store) ListJobsByState(ctx context.Context, state job.State, opts job.L jobs := make([]*job.Job, 0, len(ids)) for _, jID := range ids { var e jobEntity - if getErr := s.getEntity(ctx, jobKey(jID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.job(jID), &e); getErr != nil { continue // skip missing } if job.State(e.State) != state { @@ -360,7 +360,7 @@ func (s *Store) ListJobsByState(ctx context.Context, state job.State, opts job.L // HeartbeatJob updates the heartbeat timestamp for a running job. func (s *Store) HeartbeatJob(ctx context.Context, jobID id.JobID, _ id.WorkerID) error { - key := jobKey(jobID.String()) + key := s.keys.job(jobID.String()) var e jobEntity if err := s.getEntity(ctx, key, &e); err != nil { if isNotFound(err) { @@ -379,7 +379,7 @@ func (s *Store) HeartbeatJob(ctx context.Context, jobID id.JobID, _ id.WorkerID) func (s *Store) ReapStaleJobs(ctx context.Context, threshold time.Duration) ([]*job.Job, error) { cutoff := now().Add(-threshold) - ids, err := s.rdb.SMembers(ctx, jobIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.jobIDs()).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: reap smembers: %w", err) } @@ -387,7 +387,7 @@ func (s *Store) ReapStaleJobs(ctx context.Context, threshold time.Duration) ([]* var stale []*job.Job for _, jID := range ids { var e jobEntity - if getErr := s.getEntity(ctx, jobKey(jID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.job(jID), &e); getErr != nil { continue } if job.State(e.State) != job.StateRunning { @@ -411,14 +411,14 @@ func (s *Store) ReapStaleJobs(ctx context.Context, threshold time.Duration) ([]* // CountJobs returns the number of jobs matching the given options. func (s *Store) CountJobs(ctx context.Context, opts job.CountOpts) (int64, error) { - ids, err := s.rdb.SMembers(ctx, jobIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.jobIDs()).Result() if err != nil { return 0, fmt.Errorf("dispatch/redis: count smembers: %w", err) } var count int64 for _, jID := range ids { - raw, getErr := s.kv.GetRaw(ctx, jobKey(jID)) + raw, getErr := s.kv.GetRaw(ctx, s.keys.job(jID)) if getErr != nil { continue } diff --git a/store/redis/keys.go b/store/redis/keys.go index bcb37b3..c643765 100644 --- a/store/redis/keys.go +++ b/store/redis/keys.go @@ -2,109 +2,129 @@ package redis import "fmt" -// Redis key naming conventions for dispatch data. -// All keys are prefixed with "dispatch:" to avoid collisions. +// keys builds every Redis key and channel name the store touches. All of +// them sit under "dispatch:" so the store can share a Redis with other +// users of the same database; a tenant prefix on top of that lets several +// dispatch instances share one Redis without seeing each other's queues, +// cron locks or leadership. +type keys struct { + prefix string +} + +// base is the namespace every dispatch key lives under, tenant or not. +const base = "dispatch:" -const keyPrefix = "dispatch:" +func newKeys(prefix string) keys { return keys{prefix: prefix} } + +// full composes the key that hits Redis. The tenant prefix goes outside +// the dispatch namespace (ws_acme:dispatch:job:1) so everything a tenant +// owns shares one leading segment, the same layout the tenant's other +// Redis keys already use; an empty prefix yields the historical key. +func (k keys) full(suffix string) string { + return k.prefix + base + suffix +} // ── Job keys ── -// jobKey returns the key for a job entity: dispatch:job:{id} -func jobKey(id string) string { return keyPrefix + "job:" + id } +// job returns the key for a job entity. +func (k keys) job(id string) string { return k.full("job:" + id) } + +// queue returns the Sorted Set key for a queue. +func (k keys) queue(name string) string { return k.full("queue:" + name) } -// queueKey returns the Sorted Set key for a queue: dispatch:queue:{name} -func queueKey(name string) string { return keyPrefix + "queue:" + name } +// jobIDs is the Set tracking all job IDs for enumeration. +func (k keys) jobIDs() string { return k.full("job_ids") } -// jobIDsKey is the Set tracking all job IDs for enumeration. -const jobIDsKey = keyPrefix + "job_ids" +// wakeChannel is the pub/sub channel that announces newly enqueued jobs. +func (k keys) wakeChannel() string { return k.full("jobs:wake") } // ── Workflow keys ── -// runKey returns the key for a workflow run entity: dispatch:run:{id} -func runKey(id string) string { return keyPrefix + "run:" + id } +// run returns the key for a workflow run entity. +func (k keys) run(id string) string { return k.full("run:" + id) } -// runIDsKey is the Set tracking all run IDs for enumeration. -const runIDsKey = keyPrefix + "run_ids" +// runIDs is the Set tracking all run IDs for enumeration. +func (k keys) runIDs() string { return k.full("run_ids") } -// checkpointKey returns the key for a checkpoint: dispatch:checkpoint:{runID}:{step} -func checkpointKey(runID, step string) string { - return fmt.Sprintf("%scheckpoint:%s:%s", keyPrefix, runID, step) +// checkpoint returns the key for a checkpoint. +func (k keys) checkpoint(runID, step string) string { + return k.full(fmt.Sprintf("checkpoint:%s:%s", runID, step)) } -// checkpointIndexKey returns the Set key tracking checkpoints for a run. -func checkpointIndexKey(runID string) string { - return keyPrefix + "checkpoint_idx:" + runID +// checkpointIndex returns the Set key tracking checkpoints for a run. +func (k keys) checkpointIndex(runID string) string { + return k.full("checkpoint_idx:" + runID) } // ── Cron keys ── -// cronKey returns the key for a cron entry entity: dispatch:cron:{id} -func cronKey(id string) string { return keyPrefix + "cron:" + id } +// cron returns the key for a cron entry entity. +func (k keys) cron(id string) string { return k.full("cron:" + id) } -// cronIDsKey is the Set tracking all cron IDs for enumeration. -const cronIDsKey = keyPrefix + "cron_ids" +// cronIDs is the Set tracking all cron IDs for enumeration. +func (k keys) cronIDs() string { return k.full("cron_ids") } -// cronNamesKey maps cron names to IDs for duplicate detection. -const cronNamesKey = keyPrefix + "cron_names" +// cronNames maps cron names to IDs for duplicate detection. +func (k keys) cronNames() string { return k.full("cron_names") } // ── DLQ keys ── -// dlqKey returns the key for a DLQ entry entity: dispatch:dlq:{id} -func dlqKey(id string) string { return keyPrefix + "dlq:" + id } +// dlq returns the key for a DLQ entry entity. +func (k keys) dlq(id string) string { return k.full("dlq:" + id) } -// dlqIDsKey is the Set tracking all DLQ entry IDs for enumeration. -const dlqIDsKey = keyPrefix + "dlq_ids" +// dlqIDs is the Set tracking all DLQ entry IDs for enumeration. +func (k keys) dlqIDs() string { return k.full("dlq_ids") } // ── Event keys ── -// eventKey returns the key for an event entity: dispatch:event:{id} -func eventKey(id string) string { return keyPrefix + "event:" + id } +// event returns the key for an event entity. +func (k keys) event(id string) string { return k.full("event:" + id) } -// eventStreamKey returns the Stream key for an event name: dispatch:events:{name} -func eventStreamKey(name string) string { return keyPrefix + "events:" + name } +// eventStream returns the Stream key for an event name. +func (k keys) eventStream(name string) string { return k.full("events:" + name) } // ── Cluster keys ── -// workerKey returns the key for a worker entity: dispatch:worker:{id} -func workerKey(id string) string { return keyPrefix + "worker:" + id } +// worker returns the key for a worker entity. +func (k keys) worker(id string) string { return k.full("worker:" + id) } -// workerIDsKey is the Set tracking all worker IDs for enumeration. -const workerIDsKey = keyPrefix + "worker_ids" +// workerIDs is the Set tracking all worker IDs for enumeration. +func (k keys) workerIDs() string { return k.full("worker_ids") } -// leaderKey stores the current leader worker ID. -const leaderKey = keyPrefix + "leader" +// leader stores the current leader worker ID. +func (k keys) leader() string { return k.full("leader") } // ── Artifact keys ── -// artifactKey returns the key for an artifact entity: dispatch:artifact:{id} -func artifactKey(id string) string { return keyPrefix + "artifact:" + id } +// artifact returns the key for an artifact entity. +func (k keys) artifact(id string) string { return k.full("artifact:" + id) } -// artifactIDsKey is the Set tracking all artifact IDs for enumeration. -const artifactIDsKey = keyPrefix + "artifact_ids" +// artifactIDs is the Set tracking all artifact IDs for enumeration. +func (k keys) artifactIDs() string { return k.full("artifact_ids") } -// artifactKeyGuard maps live storage coordinates to an artifact ID. It is +// artifactGuard maps live storage coordinates to an artifact ID. It is // claimed with SETNX so concurrent creates at the same coordinates resolve // to one winner, and released on soft-delete so a purged key is reusable. -func artifactKeyGuard(backend, bucket, key string) string { - return fmt.Sprintf("%sartifact_key:%s:%s:%s", keyPrefix, backend, bucket, key) +func (k keys) artifactGuard(backend, bucket, key string) string { + return k.full(fmt.Sprintf("artifact_key:%s:%s:%s", backend, bucket, key)) } -// artifactEphemeralKey is the Sorted Set of ephemeral artifact IDs scored +// artifactEphemeral is the Sorted Set of ephemeral artifact IDs scored // by creation time. Durable artifacts are never members, which is this // backend's form of the SQL "lifecycle = 'ephemeral'" literal. -const artifactEphemeralKey = keyPrefix + "artifact_ephemeral" +func (k keys) artifactEphemeral() string { return k.full("artifact_ephemeral") } -// artifactDeletedKey is the Sorted Set of soft-deleted artifact IDs scored +// artifactDeleted is the Sorted Set of soft-deleted artifact IDs scored // by deletion time, driving the purge pass. -const artifactDeletedKey = keyPrefix + "artifact_deleted" +func (k keys) artifactDeleted() string { return k.full("artifact_deleted") } -// artifactLinksKey is the Set of link members pointing at an artifact. -func artifactLinksKey(artifactID string) string { - return keyPrefix + "artifact_links:" + artifactID +// artifactLinks is the Set of link members pointing at an artifact. +func (k keys) artifactLinks(artifactID string) string { + return k.full("artifact_links:" + artifactID) } -// ownerLinksKey is the Hash of an owner's artifact links, keyed by +// ownerLinks is the Hash of an owner's artifact links, keyed by // "name\x00attempt". -func ownerLinksKey(kind, ownerID string) string { - return fmt.Sprintf("%sartifact_owner_links:%s:%s", keyPrefix, kind, ownerID) +func (k keys) ownerLinks(kind, ownerID string) string { + return k.full(fmt.Sprintf("artifact_owner_links:%s:%s", kind, ownerID)) } diff --git a/store/redis/keys_internal_test.go b/store/redis/keys_internal_test.go new file mode 100644 index 0000000..890240e --- /dev/null +++ b/store/redis/keys_internal_test.go @@ -0,0 +1,23 @@ +package redis + +import "testing" + +func TestKeys_full(t *testing.T) { + tests := []struct { + name string + prefix string + suffix string + want string + }{ + {name: "no prefix keeps the historical key", prefix: "", suffix: "job:1", want: "dispatch:job:1"}, + {name: "tenant prefix wraps the namespace", prefix: "ws_acme:", suffix: "job:1", want: "ws_acme:dispatch:job:1"}, + {name: "wake channel follows the same rule", prefix: "ws_acme:", suffix: "jobs:wake", want: "ws_acme:dispatch:jobs:wake"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := newKeys(tt.prefix).full(tt.suffix); got != tt.want { + t.Fatalf("full(%q) with prefix %q = %q, want %q", tt.suffix, tt.prefix, got, tt.want) + } + }) + } +} diff --git a/store/redis/lease.go b/store/redis/lease.go index 762237e..d0fee6b 100644 --- a/store/redis/lease.go +++ b/store/redis/lease.go @@ -183,7 +183,7 @@ func (s *Store) RenewLease( epoch int, leaseUntil time.Time, ) error { - key := jobKey(jobID.String()) + key := s.keys.job(jobID.String()) var e jobEntity if getErr := s.getEntity(ctx, key, &e); getErr != nil { @@ -270,7 +270,7 @@ func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job t := now() - ids, err := s.rdb.SMembers(ctx, jobIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.jobIDs()).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: reclaim smembers: %w", err) } @@ -282,7 +282,7 @@ func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job } var e jobEntity - if getErr := s.getEntity(ctx, jobKey(jID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.job(jID), &e); getErr != nil { continue // gone by the time we looked } if job.State(e.State) != job.StateRunning { @@ -329,7 +329,7 @@ func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job // even for a job that was enqueued straight into running (as the // conformance suite's RunningJob helper does) and was therefore // never popped in the first place. - zErr := s.rdb.ZAdd(ctx, queueKey(after.Queue), + zErr := s.rdb.ZAdd(ctx, s.keys.queue(after.Queue), goredis.Z{Score: jobScore(after.Priority, after.RunAt), Member: jID}).Err() if zErr != nil { return nil, fmt.Errorf("dispatch/redis: reclaim requeue: %w", zErr) @@ -361,7 +361,7 @@ func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job // knows, because it wrote it. func (s *Store) claimExpired(ctx context.Context, jID string, epoch int, blob []byte) (bool, error) { res, err := reclaimScript.Run(ctx, s.rdb, - []string{jobKey(jID)}, + []string{s.keys.job(jID)}, epoch, blob, ).Int64() @@ -404,7 +404,7 @@ func (s *Store) claimExpired(ctx context.Context, jID string, epoch int, blob [] // 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 (s *Store) UpdateLeasedJob(ctx context.Context, j *job.Job, workerID id.WorkerID, epoch int) error { - key := jobKey(j.ID.String()) + key := s.keys.job(j.ID.String()) var cur jobEntity if getErr := s.getEntity(ctx, key, &cur); getErr != nil { @@ -444,7 +444,7 @@ func (s *Store) UpdateLeasedJob(ctx context.Context, j *job.Job, workerID id.Wor // pointing at a running entity is inert until dequeue's own state // check discards it. jID := j.ID.String() - qk := queueKey(next.Queue) + qk := s.keys.queue(next.Queue) runnable := job.State(next.State) == job.StatePending || job.State(next.State) == job.StateRetrying if runnable { diff --git a/store/redis/prefix_test.go b/store/redis/prefix_test.go new file mode 100644 index 0000000..5906da2 --- /dev/null +++ b/store/redis/prefix_test.go @@ -0,0 +1,165 @@ +//go:build integration + +package redis_test + +import ( + "context" + "testing" + "time" + + "github.com/xraph/dispatch" + "github.com/xraph/dispatch/cluster" + "github.com/xraph/dispatch/cron" + "github.com/xraph/dispatch/id" + "github.com/xraph/dispatch/job" + redisstore "github.com/xraph/dispatch/store/redis" +) + +// Two tenants sharing one Redis must never observe each other's dispatch +// state. Each assertion below fails on an unprefixed store because every +// key would collapse onto the same "dispatch:" namespace. + +func TestStore_KeyPrefix_jobsAreIsolated(t *testing.T) { + ctx := context.Background() + kvStore := setupTestKV(t) + a := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_a:")) + b := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_b:")) + + j := &job.Job{ + Entity: dispatch.NewEntity(), + ID: id.NewJobID(), + Name: "tenant-a-only", + Queue: "default", + Payload: []byte(`{}`), + State: job.StatePending, + MaxRetries: 3, + RunAt: time.Now().UTC(), + } + if err := a.EnqueueJob(ctx, j); err != nil { + t.Fatalf("enqueue on a: %v", err) + } + + if _, err := b.GetJob(ctx, j.ID); err == nil { + t.Fatalf("tenant b read tenant a's job %s by id", j.ID) + } + + got, err := b.DequeueJobs(ctx, job.DequeueOpts{Queues: []string{"default"}, Limit: 10}) + if err != nil { + t.Fatalf("dequeue on b: %v", err) + } + if len(got) != 0 { + t.Fatalf("tenant b dequeued %d of tenant a's jobs", len(got)) + } + + n, err := b.CountJobs(ctx, job.CountOpts{State: job.StatePending}) + if err != nil { + t.Fatalf("count on b: %v", err) + } + if n != 0 { + t.Fatalf("tenant b counted %d pending jobs, want 0", n) + } + + got, err = a.DequeueJobs(ctx, job.DequeueOpts{Queues: []string{"default"}, Limit: 10}) + if err != nil { + t.Fatalf("dequeue on a: %v", err) + } + if len(got) != 1 { + t.Fatalf("tenant a dequeued %d jobs, want its own 1", len(got)) + } +} + +func TestStore_KeyPrefix_cronNamesAreIsolated(t *testing.T) { + ctx := context.Background() + kvStore := setupTestKV(t) + a := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_a:")) + b := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_b:")) + + entry := func() *cron.Entry { + next := time.Now().Add(time.Hour).UTC() + return &cron.Entry{ + Entity: dispatch.NewEntity(), + ID: id.NewCronID(), + Name: "nightly", + Schedule: "0 0 * * *", + JobName: "rescore", + Payload: []byte(`{}`), + Enabled: true, + NextRunAt: &next, + } + } + if err := a.RegisterCron(ctx, entry()); err != nil { + t.Fatalf("register on a: %v", err) + } + // Same cron name in another tenant is a different cron, not a duplicate. + if err := b.RegisterCron(ctx, entry()); err != nil { + t.Fatalf("register on b collided with tenant a: %v", err) + } +} + +func TestStore_KeyPrefix_leadershipIsIsolated(t *testing.T) { + ctx := context.Background() + kvStore := setupTestKV(t) + a := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_a:")) + b := redisstore.New(kvStore, redisstore.WithKeyPrefix("ws_b:")) + + leaderA := registerWorker(t, a, "leader-a") + ok, err := a.AcquireLeadership(ctx, leaderA, time.Minute) + if err != nil { + t.Fatalf("acquire on a: %v", err) + } + if !ok { + t.Fatal("tenant a could not take leadership of an empty cluster") + } + + leaderB := registerWorker(t, b, "leader-b") + ok, err = b.AcquireLeadership(ctx, leaderB, time.Minute) + if err != nil { + t.Fatalf("acquire on b: %v", err) + } + if !ok { + t.Fatal("tenant b was blocked by tenant a's leader") + } +} + +// registerWorker enrols a worker on the store, since leadership can only be +// taken by a registered worker. +func registerWorker(t *testing.T, s *redisstore.Store, hostname string) id.WorkerID { + t.Helper() + w := &cluster.Worker{ + ID: id.NewWorkerID(), + Hostname: hostname, + Queues: []string{"default"}, + Concurrency: 1, + State: cluster.WorkerActive, + LastSeen: time.Now().UTC(), + CreatedAt: time.Now().UTC(), + } + if err := s.RegisterWorker(context.Background(), w); err != nil { + t.Fatalf("register worker %s: %v", hostname, err) + } + return w.ID +} + +func TestStore_KeyPrefix_emptyPrefixKeepsLegacyKeys(t *testing.T) { + ctx := context.Background() + kvStore := setupTestKV(t) + legacy := redisstore.New(kvStore) + explicit := redisstore.New(kvStore, redisstore.WithKeyPrefix("")) + + j := &job.Job{ + Entity: dispatch.NewEntity(), + ID: id.NewJobID(), + Name: "shared", + Queue: "default", + Payload: []byte(`{}`), + State: job.StatePending, + MaxRetries: 3, + RunAt: time.Now().UTC(), + } + if err := legacy.EnqueueJob(ctx, j); err != nil { + t.Fatalf("enqueue: %v", err) + } + if _, err := explicit.GetJob(ctx, j.ID); err != nil { + t.Fatalf("empty prefix must address the same keys as no prefix: %v", err) + } +} diff --git a/store/redis/store.go b/store/redis/store.go index 5e21227..6b59f7f 100644 --- a/store/redis/store.go +++ b/store/redis/store.go @@ -43,11 +43,20 @@ func WithLogger(l log.Logger) Option { return func(s *Store) { s.logger = l } } +// 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 WithKeyPrefix(prefix string) Option { + return func(s *Store) { s.keys = newKeys(prefix) } +} + // Store implements the composite store.Store interface backed by Redis // via Grove KV. type Store struct { kv *kv.Store rdb goredis.UniversalClient + keys keys logger log.Logger } @@ -57,6 +66,7 @@ func New(store *kv.Store, opts ...Option) *Store { s := &Store{ kv: store, rdb: redisdriver.UnwrapClient(store), + keys: newKeys(""), logger: log.NewNoopLogger(), } for _, o := range opts { @@ -68,6 +78,10 @@ func New(store *kv.Store, opts ...Option) *Store { // KV returns the underlying KV store. func (s *Store) KV() *kv.Store { return s.kv } +// KeyPrefix returns the tenant prefix every key is written under; empty +// when the store uses the historical unprefixed keys. +func (s *Store) KeyPrefix() string { return s.keys.prefix } + // Migrate is a no-op for Redis (schemaless). func (s *Store) Migrate(_ context.Context) error { return nil } diff --git a/store/redis/store_test.go b/store/redis/store_test.go index 274809c..3952a4d 100644 --- a/store/redis/store_test.go +++ b/store/redis/store_test.go @@ -25,8 +25,10 @@ import ( "github.com/xraph/dispatch/workflow" ) -// setupTestStore creates a Redis container and returns a connected Redis Store. -func setupTestStore(t *testing.T) *redisstore.Store { +// setupTestKV starts a Redis container and returns a flushed KV store on +// it. Tests that need more than one dispatch store on the same Redis (the +// key-prefix isolation cases) build their stores from this directly. +func setupTestKV(t *testing.T) *kv.Store { t.Helper() ctx := context.Background() @@ -61,15 +63,19 @@ func setupTestStore(t *testing.T) *redisstore.Store { _ = kvStore.Close() }) - store := redisstore.New(kvStore) - // FlushDB to start clean. client := redisdriver.UnwrapClient(kvStore) if flushErr := client.FlushDB(ctx).Err(); flushErr != nil { t.Fatalf("flush: %v", flushErr) } - return store + return kvStore +} + +// setupTestStore creates a Redis container and returns a connected Redis Store. +func setupTestStore(t *testing.T) *redisstore.Store { + t.Helper() + return redisstore.New(setupTestKV(t)) } // ────────────────────────────────────────────────── diff --git a/store/redis/wake.go b/store/redis/wake.go index 0fe8789..70b8618 100644 --- a/store/redis/wake.go +++ b/store/redis/wake.go @@ -7,10 +7,6 @@ import ( dispatchstore "github.com/xraph/dispatch/store" ) -// wakeChannel is the pub/sub channel used to signal that new jobs were -// enqueued. EnqueueJob publishes to it; StartWakeListener subscribes. -const wakeChannel = "dispatch:jobs:wake" - var _ dispatchstore.WakeNotifier = (*Store)(nil) // notifyWake signals listening instances that pending jobs exist. @@ -18,7 +14,7 @@ var _ dispatchstore.WakeNotifier = (*Store)(nil) // failed publish only costs poll latency and is not worth failing the // enqueue over. func (s *Store) notifyWake(ctx context.Context) { - _ = s.rdb.Publish(ctx, wakeChannel, "").Err() //nolint:errcheck // best-effort: polling covers missed wakes + _ = s.rdb.Publish(ctx, s.keys.wakeChannel(), "").Err() //nolint:errcheck // best-effort: polling covers missed wakes } // StartWakeListener subscribes to the dispatch wake channel and invokes @@ -30,7 +26,7 @@ func (s *Store) notifyWake(ctx context.Context) { func (s *Store) StartWakeListener(ctx context.Context, wake func()) (func(), error) { ctx, cancel := context.WithCancel(ctx) - sub := s.rdb.Subscribe(ctx, wakeChannel) + sub := s.rdb.Subscribe(ctx, s.keys.wakeChannel()) // Confirm the subscription is established so callers know push is // live before relying on it. if _, err := sub.Receive(ctx); err != nil { diff --git a/store/redis/workflow.go b/store/redis/workflow.go index 407cf1c..e4a8396 100644 --- a/store/redis/workflow.go +++ b/store/redis/workflow.go @@ -79,7 +79,7 @@ type checkpointEntity struct { // CreateRun persists a new workflow run. func (s *Store) CreateRun(ctx context.Context, run *workflow.Run) error { rID := run.ID.String() - key := runKey(rID) + key := s.keys.run(rID) exists, err := s.entityExists(ctx, key) if err != nil { @@ -94,7 +94,7 @@ func (s *Store) CreateRun(ctx context.Context, run *workflow.Run) error { return fmt.Errorf("dispatch/redis: create run set: %w", err) } - if err := s.rdb.SAdd(ctx, runIDsKey, rID).Err(); err != nil { + if err := s.rdb.SAdd(ctx, s.keys.runIDs(), rID).Err(); err != nil { return fmt.Errorf("dispatch/redis: create run index: %w", err) } return nil @@ -103,7 +103,7 @@ func (s *Store) CreateRun(ctx context.Context, run *workflow.Run) error { // GetRun retrieves a workflow run by ID. func (s *Store) GetRun(ctx context.Context, runID id.RunID) (*workflow.Run, error) { var e runEntity - if err := s.getEntity(ctx, runKey(runID.String()), &e); err != nil { + if err := s.getEntity(ctx, s.keys.run(runID.String()), &e); err != nil { if isNotFound(err) { return nil, dispatch.ErrRunNotFound } @@ -114,7 +114,7 @@ func (s *Store) GetRun(ctx context.Context, runID id.RunID) (*workflow.Run, erro // UpdateRun persists changes to an existing workflow run. func (s *Store) UpdateRun(ctx context.Context, run *workflow.Run) error { - key := runKey(run.ID.String()) + key := s.keys.run(run.ID.String()) exists, err := s.entityExists(ctx, key) if err != nil { return fmt.Errorf("dispatch/redis: update run exists: %w", err) @@ -130,7 +130,7 @@ func (s *Store) UpdateRun(ctx context.Context, run *workflow.Run) error { // ListRuns returns workflow runs matching the given options. func (s *Store) ListRuns(ctx context.Context, opts workflow.ListOpts) ([]*workflow.Run, error) { - ids, err := s.rdb.SMembers(ctx, runIDsKey).Result() + ids, err := s.rdb.SMembers(ctx, s.keys.runIDs()).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: list runs smembers: %w", err) } @@ -138,7 +138,7 @@ func (s *Store) ListRuns(ctx context.Context, opts workflow.ListOpts) ([]*workfl runs := make([]*workflow.Run, 0, len(ids)) for _, rID := range ids { var e runEntity - if getErr := s.getEntity(ctx, runKey(rID), &e); getErr != nil { + if getErr := s.getEntity(ctx, s.keys.run(rID), &e); getErr != nil { continue } if opts.State != "" && workflow.RunState(e.State) != opts.State { @@ -157,7 +157,7 @@ func (s *Store) ListRuns(ctx context.Context, opts workflow.ListOpts) ([]*workfl // SaveCheckpoint persists checkpoint data for a workflow step. func (s *Store) SaveCheckpoint(ctx context.Context, runID id.RunID, stepName string, data []byte) error { rID := runID.String() - key := checkpointKey(rID, stepName) + key := s.keys.checkpoint(rID, stepName) e := &checkpointEntity{ ID: id.NewCheckpointID().String(), @@ -171,7 +171,7 @@ func (s *Store) SaveCheckpoint(ctx context.Context, runID id.RunID, stepName str return fmt.Errorf("dispatch/redis: save checkpoint: %w", err) } - if err := s.rdb.SAdd(ctx, checkpointIndexKey(rID), stepName).Err(); err != nil { + if err := s.rdb.SAdd(ctx, s.keys.checkpointIndex(rID), stepName).Err(); err != nil { return fmt.Errorf("dispatch/redis: save checkpoint index: %w", err) } return nil @@ -179,7 +179,7 @@ func (s *Store) SaveCheckpoint(ctx context.Context, runID id.RunID, stepName str // GetCheckpoint retrieves checkpoint data for a specific workflow step. func (s *Store) GetCheckpoint(ctx context.Context, runID id.RunID, stepName string) ([]byte, error) { - key := checkpointKey(runID.String(), stepName) + key := s.keys.checkpoint(runID.String(), stepName) var e checkpointEntity if err := s.getEntity(ctx, key, &e); err != nil { if isNotFound(err) { @@ -193,14 +193,14 @@ func (s *Store) GetCheckpoint(ctx context.Context, runID id.RunID, stepName stri // ListCheckpoints returns all checkpoints for a workflow run. func (s *Store) ListCheckpoints(ctx context.Context, runID id.RunID) ([]*workflow.Checkpoint, error) { rID := runID.String() - steps, err := s.rdb.SMembers(ctx, checkpointIndexKey(rID)).Result() + steps, err := s.rdb.SMembers(ctx, s.keys.checkpointIndex(rID)).Result() if err != nil { return nil, fmt.Errorf("dispatch/redis: list checkpoints: %w", err) } checkpoints := make([]*workflow.Checkpoint, 0, len(steps)) for _, step := range steps { - key := checkpointKey(rID, step) + key := s.keys.checkpoint(rID, step) var e checkpointEntity if getErr := s.getEntity(ctx, key, &e); getErr != nil { continue @@ -244,7 +244,7 @@ func (s *Store) DeleteCheckpointsAfter(ctx context.Context, runID id.RunID, afte // Get the target checkpoint's time. var target checkpointEntity - if err := s.getEntity(ctx, checkpointKey(rID, afterStep), &target); err != nil { + if err := s.getEntity(ctx, s.keys.checkpoint(rID, afterStep), &target); err != nil { if isNotFound(err) { return nil // step not found; nothing to delete } @@ -252,13 +252,13 @@ func (s *Store) DeleteCheckpointsAfter(ctx context.Context, runID id.RunID, afte } // List all step names for this run. - steps, err := s.rdb.SMembers(ctx, checkpointIndexKey(rID)).Result() + steps, err := s.rdb.SMembers(ctx, s.keys.checkpointIndex(rID)).Result() if err != nil { return fmt.Errorf("dispatch/redis: list checkpoint steps: %w", err) } for _, step := range steps { - key := checkpointKey(rID, step) + key := s.keys.checkpoint(rID, step) var e checkpointEntity if getErr := s.getEntity(ctx, key, &e); getErr != nil { continue @@ -267,7 +267,7 @@ func (s *Store) DeleteCheckpointsAfter(ctx context.Context, runID id.RunID, afte if delErr := s.rdb.Del(ctx, key).Err(); delErr != nil { return fmt.Errorf("delete checkpoint %s: %w", key, delErr) } - if remErr := s.rdb.SRem(ctx, checkpointIndexKey(rID), step).Err(); remErr != nil { + if remErr := s.rdb.SRem(ctx, s.keys.checkpointIndex(rID), step).Err(); remErr != nil { return fmt.Errorf("remove checkpoint index %s: %w", step, remErr) } }