From 12f5d5f8ede6288671e7f408eb4add2684c461ef Mon Sep 17 00:00:00 2001 From: CoffeeKnyte <67730400+CoffeeKnyte@users.noreply.github.com> Date: Fri, 31 Jul 2026 00:22:23 +0000 Subject: [PATCH] fix(playback): share the live-stream picture across replicas MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every API replica ran its own enforcer over only its own SessionManager plus the edge Redis records, because `nodesessions.NewTracker` is constructed only in proxy/transcode mode — integrated streams never reached the shared `silo:sessions:` namespace. Replicas were mutually blind, so under-enforcement of the concurrent-stream cap was certain, and two replicas trimming from different snapshots could over-kill. Per decision A6 (Option A): - `nodesessions.Publisher` mirrors the integrated process's live sessions into the shared keyspace every 10s. It re-SETs every live record on every tick so the 60s record TTL is renewed, and diffs only deletions so a stopped session leaves the shared picture at once rather than lingering for the TTL. - The publisher's key namespace is derived from a process-unique instance id, not `resolveNodeIdentity()`. That helper returns SILO_NODE_NAME/NODE_NAME/ hostname, so an operator setting it in shared env and scaling the deployment would give every replica the same namespace and they would delete each other's session records — worse than the blindness this fixes. - `streamenforcer.Coordinator` elects a single evaluator per tick via a renewable Redis lease (a plain SET NX would lock the holder out of its own next tick). Each pass is bounded by the evaluation interval so it can never outlive the lease. No coordinator, or a Redis error, evaluates anyway: failing to coordinate must never mean failing to enforce. Also closes the cross-replica gap Batch 3 recorded as an accepted limitation: `mirrorToRedis` was an unconditional SET, so two replicas revoking the same key could lose the stronger kill — and edges learn kills only from Redis. It now merges server-side in Lua with the same two independent monotonic dimensions as `applyLocal` (expiry never earlier, cutoff never backward, reason follows the newer cutoff), sharing one Go definition of that comparison. Two details that are easy to get wrong: - The merge compares exact (unix_sec, nsec) pairs carried as additive fields on a dedicated mirror payload. RFC3339 strings cannot be compared lexicographically (Go omits trailing zeros, so "…:00Z" sorts after "…:00.5Z"), and millisecond truncation could retain an older cutoff and let a credential issued between two same-millisecond revocations keep streaming. - The script merges before deciding to delete. Deleting on a lapsed *incoming* revocation, as the old unconditional path did, could remove a live permanent or longer-lived kill written by another replica. The mirror payload is additive, so old and new nodes interoperate during a rolling deploy, and any script failure falls back to the previous plain SET with a once-only warning. This fixes cross-replica *monitoring* only. Synchronous admission remains per-process, so streams spread across replicas are still all admitted and are trimmed asynchronously by the enforcer rather than refused at the door. A distributed admission gate is separate work. Part 2 of 3 for the Batch 4 liveness/replica work. Part of #305 --- cmd/silo/main.go | 20 ++- internal/nodesessions/publisher.go | 124 +++++++++++++ internal/nodesessions/publisher_test.go | 146 +++++++++++++++ internal/nodesessions/tracker.go | 14 +- internal/streamenforcer/coordinator.go | 58 ++++++ internal/streamenforcer/enforcer.go | 22 ++- internal/streamenforcer/enforcer_test.go | 158 ++++++++++++++++ internal/streamrevoke/store.go | 141 +++++++++++++-- internal/streamrevoke/store_redis_test.go | 208 ++++++++++++++++++++++ internal/streamrevoke/store_test.go | 30 ++++ 10 files changed, 896 insertions(+), 25 deletions(-) create mode 100644 internal/nodesessions/publisher.go create mode 100644 internal/nodesessions/publisher_test.go create mode 100644 internal/streamenforcer/coordinator.go create mode 100644 internal/streamrevoke/store_redis_test.go diff --git a/cmd/silo/main.go b/cmd/silo/main.go index c0c63bf9..1caded2c 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -1708,6 +1708,7 @@ func main() { // Build the reconciler early enough that playback handlers can trigger // immediate session syncs after start/stop events. nodeIdentity := resolveNodeIdentity() + instanceID := nodeIdentity + ":" + uuid.NewString() var reconciler *worker.Reconciler var heartbeatWriter *worker.HeartbeatWriter @@ -1752,6 +1753,15 @@ func main() { }) var monitorSource streammonitor.Source = localSource if apiRedisClient != nil { + publisher := nodesessions.NewPublisher(apiRedisClient, instanceID, nodeIdentity) + publisher.Start(appCtx, 10*time.Second, func() []nodesessions.SessionInfo { + return streammonitor.LiveLocalSessions(sessionMgr, nodeIdentity) + }) + defer func() { + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cleanupCancel() + publisher.Cleanup(cleanupCtx) + }() monitorSource = streammonitor.NewMultiSource(localSource, streammonitor.NewRedisSource(apiRedisClient)) } // Resolve caps through the session manager so the enforcer and synchronous @@ -1768,13 +1778,19 @@ func main() { "error", ttlErr) overCapRevocationTTL = streamenforcer.DefaultRevocationTTL } - streamenforcer.NewWithRevocationTTL(monitorSource, func(ctx context.Context, userID int) (int, error) { + enforcer := streamenforcer.NewWithRevocationTTL(monitorSource, func(ctx context.Context, userID int) (int, error) { limits, err := sessionMgr.EffectiveLimits(ctx, userID) if err != nil { return 0, err } return limits.MaxStreams, nil - }, streamRevocation, 0, overCapRevocationTTL).Start(appCtx) + }, streamRevocation, 0, overCapRevocationTTL) + if apiRedisClient != nil { + enforcer.SetCoordinator(streamenforcer.NewRedisCoordinator( + apiRedisClient, instanceID, streamenforcer.DefaultInterval, + )) + } + enforcer.Start(appCtx) nodeURL := fmt.Sprintf("http://%s%s", nodeIdentity, cfg.Server.Listen) heartbeatWriter = worker.NewHeartbeatWriter(deps.DB, nodeIdentity, mode, nodeURL) diff --git a/internal/nodesessions/publisher.go b/internal/nodesessions/publisher.go new file mode 100644 index 00000000..c280cfd6 --- /dev/null +++ b/internal/nodesessions/publisher.go @@ -0,0 +1,124 @@ +package nodesessions + +import ( + "context" + "encoding/json" + "log/slog" + "sync" + "time" + + "github.com/redis/go-redis/v9" +) + +// Publisher mirrors the integrated API process's live sessions into the +// shared node-session keyspace. +type Publisher struct { + rdb *redis.Client + nodeName string + nodeHash string + ttl time.Duration + + mu sync.Mutex + ids map[string]struct{} +} + +// NewPublisher creates a publisher whose key namespace is derived from the +// process-unique instanceID. nodeName remains display metadata only. +func NewPublisher(rdb *redis.Client, instanceID, nodeName string) *Publisher { + return &Publisher{ + rdb: rdb, + nodeName: nodeName, + nodeHash: namespaceHash(instanceID), + ttl: sessionTTL, + ids: make(map[string]struct{}), + } +} + +// Publish refreshes every currently-live record and immediately removes only +// records which disappeared since the preceding successful snapshot. +func (p *Publisher) Publish(ctx context.Context, infos []SessionInfo) { + if p == nil || p.rdb == nil { + return + } + + type record struct { + id string + data []byte + } + records := make([]record, 0, len(infos)) + current := make(map[string]struct{}, len(infos)) + for _, info := range infos { + if info.SessionID == "" { + continue + } + info.NodeName = p.nodeName + data, err := json.Marshal(info) + if err != nil { + slog.DebugContext(ctx, "integrated session marshal failed", + "component", "nodesessions", "session", info.SessionID, "error", err) + continue + } + records = append(records, record{id: info.SessionID, data: data}) + current[info.SessionID] = struct{}{} + } + + p.mu.Lock() + defer p.mu.Unlock() + + pipe := p.rdb.Pipeline() + for _, rec := range records { + pipe.Set(ctx, sessionRedisKey(p.nodeHash, rec.id), rec.data, p.ttl) + } + for id := range p.ids { + if _, live := current[id]; !live { + pipe.Del(ctx, sessionRedisKey(p.nodeHash, id)) + } + } + if _, err := pipe.Exec(ctx); err != nil { + slog.DebugContext(ctx, "integrated session publish failed", + "component", "nodesessions", "error", err) + return + } + p.ids = current +} + +// Start publishes snapshots at interval until ctx is canceled. +func (p *Publisher) Start(ctx context.Context, interval time.Duration, snapshot func() []SessionInfo) { + if p == nil || p.rdb == nil || interval <= 0 || snapshot == nil { + return + } + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + p.Publish(ctx, snapshot()) + } + } + }() +} + +// Cleanup removes all records owned by this process. +func (p *Publisher) Cleanup(ctx context.Context) { + if p == nil || p.rdb == nil { + return + } + p.mu.Lock() + defer p.mu.Unlock() + if len(p.ids) == 0 { + return + } + pipe := p.rdb.Pipeline() + for id := range p.ids { + pipe.Del(ctx, sessionRedisKey(p.nodeHash, id)) + } + if _, err := pipe.Exec(ctx); err != nil { + slog.DebugContext(ctx, "integrated session cleanup failed", + "component", "nodesessions", "error", err) + return + } + p.ids = make(map[string]struct{}) +} diff --git a/internal/nodesessions/publisher_test.go b/internal/nodesessions/publisher_test.go new file mode 100644 index 00000000..50fb8b30 --- /dev/null +++ b/internal/nodesessions/publisher_test.go @@ -0,0 +1,146 @@ +package nodesessions + +import ( + "context" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/redis/go-redis/v9" +) + +type expiringRedisHook struct { + mu sync.Mutex + expires map[string]time.Time +} + +func (h *expiringRedisHook) DialHook(next redis.DialHook) redis.DialHook { + return next +} + +func (h *expiringRedisHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook { + return func(ctx context.Context, cmd redis.Cmder) error { + return h.process(ctx, cmd) + } +} + +func (h *expiringRedisHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook { + return func(ctx context.Context, cmds []redis.Cmder) error { + for _, cmd := range cmds { + if err := h.process(ctx, cmd); err != nil { + return err + } + } + return nil + } +} + +func (h *expiringRedisHook) process(_ context.Context, cmd redis.Cmder) error { + args := cmd.Args() + if len(args) < 2 { + return nil + } + key, _ := args[1].(string) + h.mu.Lock() + defer h.mu.Unlock() + switch strings.ToLower(cmd.Name()) { + case "set": + expiry := time.Time{} + for i := 3; i+1 < len(args); i++ { + if strings.EqualFold(strings.TrimSpace(toString(args[i])), "px") { + ms, _ := strconv.ParseInt(toString(args[i+1]), 10, 64) + expiry = time.Now().Add(time.Duration(ms) * time.Millisecond) + break + } + } + h.expires[key] = expiry + case "del": + delete(h.expires, key) + } + return nil +} + +func toString(value any) string { + switch value := value.(type) { + case string: + return value + case []byte: + return string(value) + case int: + return strconv.Itoa(value) + case int64: + return strconv.FormatInt(value, 10) + case time.Duration: + return strconv.FormatInt(value.Milliseconds(), 10) + default: + return "" + } +} + +func (h *expiringRedisHook) has(key string) bool { + h.mu.Lock() + defer h.mu.Unlock() + expiry, ok := h.expires[key] + if ok && !expiry.IsZero() && !time.Now().Before(expiry) { + delete(h.expires, key) + return false + } + return ok +} + +func testPublisherRedis(t *testing.T) (*redis.Client, *expiringRedisHook) { + t.Helper() + hook := &expiringRedisHook{expires: make(map[string]time.Time)} + rdb := redis.NewClient(&redis.Options{Addr: "unused.invalid:6379"}) + rdb.AddHook(hook) + t.Cleanup(func() { _ = rdb.Close() }) + return rdb, hook +} + +func TestPublishersWithSharedNodeNameDoNotClobberEachOther(t *testing.T) { + ctx := context.Background() + rdb, redisState := testPublisherRedis(t) + suffix := time.Now().UTC().Format("20060102150405.000000000") + first := NewPublisher(rdb, "first:"+suffix, "shared-name") + second := NewPublisher(rdb, "second:"+suffix, "shared-name") + t.Cleanup(func() { + first.Cleanup(ctx) + second.Cleanup(ctx) + }) + + info := SessionInfo{SessionID: "same-session"} + first.Publish(ctx, []SessionInfo{info}) + second.Publish(ctx, []SessionInfo{info}) + first.Publish(ctx, nil) + + firstKey := sessionRedisKey(first.nodeHash, info.SessionID) + secondKey := sessionRedisKey(second.nodeHash, info.SessionID) + if redisState.has(firstKey) { + t.Fatal("departed first key is still present") + } + if !redisState.has(secondKey) { + t.Fatal("live second key is absent") + } +} + +func TestPublisherRefreshesStableRecordPastTTL(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + rdb, redisState := testPublisherRedis(t) + suffix := time.Now().UTC().Format("20060102150405.000000000") + publisher := NewPublisher(rdb, "refresh:"+suffix, "node") + publisher.ttl = 80 * time.Millisecond + t.Cleanup(func() { publisher.Cleanup(context.Background()) }) + + publisher.Start(ctx, 20*time.Millisecond, func() []SessionInfo { + return []SessionInfo{{SessionID: "stable"}} + }) + time.Sleep(220 * time.Millisecond) + + key := sessionRedisKey(publisher.nodeHash, "stable") + if !redisState.has(key) { + t.Fatal("stable key is absent after > TTL") + } +} diff --git a/internal/nodesessions/tracker.go b/internal/nodesessions/tracker.go index 398d31ce..c00a5e2a 100644 --- a/internal/nodesessions/tracker.go +++ b/internal/nodesessions/tracker.go @@ -108,13 +108,12 @@ type Tracker struct { // NewTracker creates a session tracker for the given node. // rdb may be nil, in which case all operations are no-ops. func NewTracker(rdb *redis.Client, nodeURL, nodeName, nodeType string) *Tracker { - h := sha256.Sum256([]byte(nodeURL)) return &Tracker{ rdb: rdb, nodeURL: nodeURL, nodeName: nodeName, nodeType: nodeType, - nodeHash: hex.EncodeToString(h[:4]), // 8 hex chars + nodeHash: namespaceHash(nodeURL), sessions: make(map[string]sessionRefs), ephemeral: make(map[string]time.Time), touched: make(map[string]time.Time), @@ -125,7 +124,16 @@ func NewTracker(rdb *redis.Client, nodeURL, nodeName, nodeType string) *Tracker // redisKey returns the full Redis key for a session. func (tr *Tracker) redisKey(sessionID string) string { - return KeyPrefix + tr.nodeHash + ":" + sessionID + return sessionRedisKey(tr.nodeHash, sessionID) +} + +func namespaceHash(namespace string) string { + h := sha256.Sum256([]byte(namespace)) + return hex.EncodeToString(h[:4]) +} + +func sessionRedisKey(nodeHash, sessionID string) string { + return KeyPrefix + nodeHash + ":" + sessionID } // NodeHash returns the node's hash prefix used in Redis keys. diff --git a/internal/streamenforcer/coordinator.go b/internal/streamenforcer/coordinator.go new file mode 100644 index 00000000..f244e667 --- /dev/null +++ b/internal/streamenforcer/coordinator.go @@ -0,0 +1,58 @@ +package streamenforcer + +import ( + "context" + "time" + + "github.com/redis/go-redis/v9" +) + +const coordinatorKey = "silo:enforcer:leader" + +var acquireLeaseScript = redis.NewScript(` +local v = redis.call('GET', KEYS[1]) +if v == false or v == ARGV[1] then + redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2]) + return 1 +end +return 0 +`) + +// Coordinator elects one enforcer replica for an evaluation pass. +type Coordinator interface { + Acquire(ctx context.Context) (bool, error) +} + +// RedisCoordinator uses a renewable lease to select one evaluator. +type RedisCoordinator struct { + rdb *redis.Client + instanceID string + leaseTTL time.Duration +} + +// NewRedisCoordinator creates a coordinator whose lease lasts twice the +// evaluation interval. +func NewRedisCoordinator(rdb *redis.Client, instanceID string, interval time.Duration) *RedisCoordinator { + if interval <= 0 { + interval = DefaultInterval + } + return &RedisCoordinator{rdb: rdb, instanceID: instanceID, leaseTTL: 2 * interval} +} + +// Acquire claims or renews this instance's lease. +func (c *RedisCoordinator) Acquire(ctx context.Context) (bool, error) { + if c == nil || c.rdb == nil { + return true, nil + } + result, err := acquireLeaseScript.Run( + ctx, + c.rdb, + []string{coordinatorKey}, + c.instanceID, + c.leaseTTL.Milliseconds(), + ).Int() + if err != nil { + return false, err + } + return result == 1, nil +} diff --git a/internal/streamenforcer/enforcer.go b/internal/streamenforcer/enforcer.go index f9039f56..1b152e57 100644 --- a/internal/streamenforcer/enforcer.go +++ b/internal/streamenforcer/enforcer.go @@ -60,6 +60,15 @@ type Enforcer struct { interval time.Duration revocationTTL time.Duration now func() time.Time + coordinator Coordinator +} + +// SetCoordinator installs the optional cross-replica evaluator coordinator. +// Without one, every tick evaluates as before. +func (e *Enforcer) SetCoordinator(c Coordinator) { + if e != nil { + e.coordinator = c + } } // New builds an enforcer. interval <= 0 uses DefaultInterval. @@ -131,7 +140,18 @@ func (e *Enforcer) Start(ctx context.Context) { // evaluate runs one pass: snapshot → per-user over-cap check → revoke victims. // Exported behavior is covered by EvaluateOnce for tests. func (e *Enforcer) evaluate(ctx context.Context) { - if err := e.EvaluateOnce(ctx); err != nil { + if e.coordinator != nil { + acquired, err := e.coordinator.Acquire(ctx) + if err != nil { + // Coordination failure must not disable enforcement. + slog.DebugContext(ctx, "stream enforcer coordination failed; evaluating anyway", "error", err) + } else if !acquired { + return + } + } + passCtx, cancel := context.WithTimeout(ctx, e.interval) + defer cancel() + if err := e.EvaluateOnce(passCtx); err != nil { slog.DebugContext(ctx, "stream enforcer evaluate failed", "error", err) } } diff --git a/internal/streamenforcer/enforcer_test.go b/internal/streamenforcer/enforcer_test.go index e8d590a8..f9f77e37 100644 --- a/internal/streamenforcer/enforcer_test.go +++ b/internal/streamenforcer/enforcer_test.go @@ -3,13 +3,16 @@ package streamenforcer import ( "context" "errors" + "os" "sort" + "sync" "testing" "time" "github.com/Silo-Server/silo-server/internal/nodesessions" "github.com/Silo-Server/silo-server/internal/playback" "github.com/Silo-Server/silo-server/internal/streammonitor" + "github.com/redis/go-redis/v9" ) type fakeSource struct { @@ -246,3 +249,158 @@ func TestParseRevocationTTL(t *testing.T) { } } } + +type firstCoordinator struct { + mu sync.Mutex + acquired bool +} + +func (c *firstCoordinator) Acquire(context.Context) (bool, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.acquired { + return false, nil + } + c.acquired = true + return true, nil +} + +type errorCoordinator struct{} + +func (errorCoordinator) Acquire(context.Context) (bool, error) { + return false, errors.New("redis unavailable") +} + +type countingSource struct { + mu sync.Mutex + snap streammonitor.Snapshot + calls int +} + +func (s *countingSource) Snapshot(context.Context) (streammonitor.Snapshot, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.calls++ + return s.snap, nil +} + +func TestSharedCoordinatorAllowsExactlyOneEnforcerEvaluation(t *testing.T) { + base := time.Now() + source := &countingSource{snap: streammonitor.Snapshot{Streams: []streammonitor.LiveStream{ + stream("keep", 7, base.Add(time.Minute)), + stream("victim", 7, base), + }}} + revoker := &fakeRevoker{} + coordinator := &firstCoordinator{} + enforcers := []*Enforcer{ + New(source, func(context.Context, int) (int, error) { return 1, nil }, revoker, time.Second), + New(source, func(context.Context, int) (int, error) { return 1, nil }, revoker, time.Second), + } + for _, enforcer := range enforcers { + enforcer.SetCoordinator(coordinator) + } + + var wg sync.WaitGroup + for _, enforcer := range enforcers { + wg.Add(1) + go func() { + defer wg.Done() + enforcer.evaluate(context.Background()) + }() + } + wg.Wait() + source.mu.Lock() + snapshotCalls := source.calls + source.mu.Unlock() + if snapshotCalls != 1 { + t.Fatalf("snapshot calls = %d, want exactly 1", snapshotCalls) + } + if len(revoker.revoked) != 1 || revoker.revoked[0] != "victim" { + t.Fatalf("revoked = %v, want exactly [victim]", revoker.revoked) + } +} + +func TestCoordinatorErrorStillEvaluates(t *testing.T) { + base := time.Now() + revoker := &fakeRevoker{} + enforcer := New(fakeSource{snap: streammonitor.Snapshot{Streams: []streammonitor.LiveStream{ + stream("keep", 7, base.Add(time.Minute)), + stream("victim", 7, base), + }}}, func(context.Context, int) (int, error) { return 1, nil }, revoker, time.Second) + enforcer.SetCoordinator(errorCoordinator{}) + enforcer.evaluate(context.Background()) + if len(revoker.revoked) != 1 || revoker.revoked[0] != "victim" { + t.Fatalf("revoked = %v, want [victim]", revoker.revoked) + } +} + +type blockingSource struct { + deadlineObserved chan struct{} +} + +func (s blockingSource) Snapshot(ctx context.Context) (streammonitor.Snapshot, error) { + if _, ok := ctx.Deadline(); !ok { + close(s.deadlineObserved) + return streammonitor.Snapshot{}, errors.New("evaluation context has no deadline") + } + <-ctx.Done() + close(s.deadlineObserved) + return streammonitor.Snapshot{}, ctx.Err() +} + +func TestEvaluationPassIsBoundedByInterval(t *testing.T) { + observed := make(chan struct{}) + enforcer := New( + blockingSource{deadlineObserved: observed}, + func(context.Context, int) (int, error) { return 1, nil }, + &fakeRevoker{}, + 25*time.Millisecond, + ) + start := time.Now() + enforcer.evaluate(context.Background()) + elapsed := time.Since(start) + select { + case <-observed: + default: + t.Fatal("source did not observe the evaluation deadline") + } + if elapsed < 20*time.Millisecond || elapsed > 250*time.Millisecond { + t.Fatalf("evaluation elapsed %v, want bounded near 25ms", elapsed) + } +} + +func TestRedisCoordinatorElectsOneHolderAndRenews(t *testing.T) { + addr := os.Getenv("SILO_TEST_REDIS_ADDR") + if addr == "" { + t.Skip("SILO_TEST_REDIS_ADDR is not set") + } + ctx := context.Background() + rdb := redis.NewClient(&redis.Options{Addr: addr}) + t.Cleanup(func() { + _ = rdb.Del(ctx, coordinatorKey).Err() + _ = rdb.Close() + }) + if err := rdb.Ping(ctx).Err(); err != nil { + t.Fatalf("ping test Redis: %v", err) + } + if err := rdb.Del(ctx, coordinatorKey).Err(); err != nil { + t.Fatal(err) + } + + const interval = 80 * time.Millisecond + first := NewRedisCoordinator(rdb, "first", interval) + second := NewRedisCoordinator(rdb, "second", interval) + if acquired, err := first.Acquire(ctx); err != nil || !acquired { + t.Fatalf("first acquire=%v err=%v, want true", acquired, err) + } + if acquired, err := second.Acquire(ctx); err != nil || acquired { + t.Fatalf("second acquire=%v err=%v, want false", acquired, err) + } + time.Sleep(90 * time.Millisecond) + if acquired, err := first.Acquire(ctx); err != nil || !acquired { + t.Fatalf("renew acquire=%v err=%v, want true", acquired, err) + } + if ttl, err := rdb.PTTL(ctx, coordinatorKey).Result(); err != nil || ttl < 120*time.Millisecond { + t.Fatalf("renewed lease TTL=%v err=%v, want >=120ms", ttl, err) + } +} diff --git a/internal/streamrevoke/store.go b/internal/streamrevoke/store.go index b92a61c8..6ee0b2ed 100644 --- a/internal/streamrevoke/store.go +++ b/internal/streamrevoke/store.go @@ -51,6 +51,67 @@ const ( defaultTTL = 24 * time.Hour ) +var mergeMirrorScript = redis.NewScript(` +local incoming = cjson.decode(ARGV[1]) +local existing_raw = redis.call('GET', KEYS[1]) +local merged = incoming + +if existing_raw ~= false then + local ok, existing = pcall(cjson.decode, existing_raw) + if ok + and type(existing.revoked_at_sec) == 'number' + and type(existing.revoked_at_nsec) == 'number' + and type(existing.expires_at_sec) == 'number' + and type(existing.expires_at_nsec) == 'number' then + merged = existing + + local incoming_permanent = incoming.expires_at_sec == 0 + local existing_permanent = existing.expires_at_sec == 0 + local incoming_expiry_later = incoming.expires_at_sec > existing.expires_at_sec + or (incoming.expires_at_sec == existing.expires_at_sec + and incoming.expires_at_nsec > existing.expires_at_nsec) + if incoming_permanent or (not existing_permanent and incoming_expiry_later) then + merged.expires_at = incoming.expires_at + merged.expires_at_sec = incoming.expires_at_sec + merged.expires_at_nsec = incoming.expires_at_nsec + end + + local incoming_cutoff_later = incoming.revoked_at_sec > existing.revoked_at_sec + or (incoming.revoked_at_sec == existing.revoked_at_sec + and incoming.revoked_at_nsec > existing.revoked_at_nsec) + if incoming_cutoff_later then + merged.revoked_at = incoming.revoked_at + merged.revoked_at_sec = incoming.revoked_at_sec + merged.revoked_at_nsec = incoming.revoked_at_nsec + merged.reason = incoming.reason + end + end +end + +local permanent = merged.expires_at_sec == 0 +local expired = not permanent and ( + merged.expires_at_sec < tonumber(ARGV[2]) + or (merged.expires_at_sec == tonumber(ARGV[2]) + and merged.expires_at_nsec <= tonumber(ARGV[3])) +) +if expired then + redis.call('DEL', KEYS[1]) + return 0 +end + +local encoded = cjson.encode(merged) +if permanent then + redis.call('SET', KEYS[1], encoded) +else + local ttl_ms = math.ceil( + (merged.expires_at_sec - tonumber(ARGV[2])) * 1000 + + (merged.expires_at_nsec - tonumber(ARGV[3])) / 1000000 + ) + redis.call('SET', KEYS[1], encoded, 'PX', math.max(1, ttl_ms)) +end +return 1 +`) + // Kind identifies what a revocation targets. type Kind string @@ -143,6 +204,7 @@ type Store struct { items map[Key]Revocation tombstones map[Key]time.Time tombstoneExpires map[Key]time.Time + mirrorWarnOnce sync.Once } // New builds a Store, applying defaults for any unset Options. @@ -311,6 +373,19 @@ func (r Revocation) effectiveExpiry() time.Time { return r.ExpiresAt } +// mergeRevocations is the Go definition of the two independent monotonic +// dimensions also applied atomically by mergeMirrorScript. +func mergeRevocations(existing, incoming Revocation) Revocation { + if incoming.effectiveExpiry().After(existing.effectiveExpiry()) { + existing.ExpiresAt = incoming.ExpiresAt + } + if incoming.RevokedAt.After(existing.RevokedAt) { + existing.RevokedAt = incoming.RevokedAt + existing.Reason = incoming.Reason + } + return existing +} + // applyLocal independently merges a revocation's two monotonic dimensions: // expiry never moves earlier, while the user-cutoff RevokedAt never moves // backward. Reason follows the record that supplied the newer cutoff. Keeping @@ -331,14 +406,7 @@ func (s *Store) applyLocal(r Revocation) { } } if existing, ok := s.items[r.key()]; ok { - if r.effectiveExpiry().After(existing.effectiveExpiry()) { - existing.ExpiresAt = r.ExpiresAt - } - if r.RevokedAt.After(existing.RevokedAt) { - existing.RevokedAt = r.RevokedAt - existing.Reason = r.Reason - } - s.items[r.key()] = existing + s.items[r.key()] = mergeRevocations(existing, r) } else { s.items[r.key()] = r } @@ -405,10 +473,8 @@ func (s *Store) revokeWithWarningsLocked(ctx context.Context, key Key, reason st merged := s.effective(r.key()) // Redis mirror for multi-node propagation to edges. Mirror the merged copy - // applyLocal kept. This SET is deliberately serialized by opMu in this - // process. Separate central replicas can still race and overwrite a newer - // value because Redis does not atomically merge records; that is deferred to - // the multi-replica work, not addressed by narrowing this lock. + // applyLocal kept; Redis atomically merges it with writes from other central + // replicas. if err := s.mirrorToRedis(ctx, merged); err != nil { warnings = append(warnings, "Redis mirror was not updated") } @@ -442,8 +508,29 @@ func (s *Store) revokeWithWarningsLocked(ctx context.Context, key Key, reason st return warnings, nil } -// mirrorToRedis writes (or deletes) the Redis mirror key for a revocation with a -// TTL of the time remaining until r.ExpiresAt. It is the single shared +type mirrorPayload struct { + Revocation + RevokedAtSec int64 `json:"revoked_at_sec"` + RevokedAtNSec int64 `json:"revoked_at_nsec"` + ExpiresAtSec int64 `json:"expires_at_sec"` + ExpiresAtNSec int64 `json:"expires_at_nsec"` +} + +func newMirrorPayload(r Revocation) mirrorPayload { + payload := mirrorPayload{ + Revocation: r, + RevokedAtSec: r.RevokedAt.Unix(), + RevokedAtNSec: int64(r.RevokedAt.Nanosecond()), + } + if !r.ExpiresAt.IsZero() { + payload.ExpiresAtSec = r.ExpiresAt.Unix() + payload.ExpiresAtNSec = int64(r.ExpiresAt.Nanosecond()) + } + return payload +} + +// mirrorToRedis atomically merges the Redis mirror with a revocation and sets +// its TTL to the time remaining until the merged expiry. It is the single shared // Redis-arming path, used both by Revoke and by the durable warm loops: edge // nodes (proxy/transcode) have no durable store and learn kills ONLY from // Redis, so re-arming Redis from the durable source of truth after a Redis flush @@ -454,15 +541,32 @@ func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) error { if s.rdb == nil { return nil } - data, err := json.Marshal(r) + data, err := json.Marshal(newMirrorPayload(r)) if err != nil { slog.WarnContext(ctx, "streamrevoke marshal failed", "error", err, "kind", r.Kind, "id", r.ID) return err } + now := time.Now() + if err := mergeMirrorScript.Run( + ctx, + s.rdb, + []string{redisKey(r.key())}, + data, + now.Unix(), + now.Nanosecond(), + ).Err(); err == nil { + return nil + } else { + s.mirrorWarnOnce.Do(func() { + slog.WarnContext(ctx, "streamrevoke atomic Redis merge unavailable; falling back to plain mirror writes", + "error", err) + }) + } + return s.mirrorToRedisFallback(ctx, r, data) +} + +func (s *Store) mirrorToRedisFallback(ctx context.Context, r Revocation, data []byte) error { if r.ExpiresAt.IsZero() { - // Permanent kill: set without a TTL, matching applyLocal/effectiveExpiry - // treating a zero ExpiresAt as "never". A TTL of time.Until(zero) would be - // hugely negative and drop the key, so late-joining edges would miss it. if err := s.rdb.Set(ctx, redisKey(r.key()), data, 0).Err(); err != nil { slog.WarnContext(ctx, "streamrevoke redis set failed", "error", err, "kind", r.Kind, "id", r.ID) return err @@ -471,7 +575,6 @@ func (s *Store) mirrorToRedis(ctx context.Context, r Revocation) error { } ttl := time.Until(r.ExpiresAt) if ttl <= 0 { - // Already expired: nothing to mirror in Redis. if err := s.rdb.Del(ctx, redisKey(r.key())).Err(); err != nil { slog.DebugContext(ctx, "streamrevoke redis del failed", "error", err, "kind", r.Kind, "id", r.ID) return err diff --git a/internal/streamrevoke/store_redis_test.go b/internal/streamrevoke/store_redis_test.go new file mode 100644 index 00000000..3590783e --- /dev/null +++ b/internal/streamrevoke/store_redis_test.go @@ -0,0 +1,208 @@ +package streamrevoke + +import ( + "context" + "encoding/json" + "os" + "sync" + "testing" + "time" + + "github.com/redis/go-redis/v9" +) + +func realRedis(t *testing.T) *redis.Client { + t.Helper() + addr := os.Getenv("SILO_TEST_REDIS_ADDR") + if addr == "" { + t.Skip("SILO_TEST_REDIS_ADDR is not set") + } + rdb := redis.NewClient(&redis.Options{Addr: addr}) + t.Cleanup(func() { _ = rdb.Close() }) + if err := rdb.Ping(context.Background()).Err(); err != nil { + t.Fatalf("ping test Redis: %v", err) + } + return rdb +} + +func uniqueRedisRevocation(t *testing.T) (Key, string) { + t.Helper() + id := t.Name() + ":" + time.Now().UTC().Format("20060102150405.000000000") + key := Key{Kind: KindSession, ID: id} + return key, redisKey(key) +} + +func readMirror(t *testing.T, rdb *redis.Client, key string) mirrorPayload { + t.Helper() + data, err := rdb.Get(context.Background(), key).Bytes() + if err != nil { + t.Fatalf("get mirror: %v", err) + } + var payload mirrorPayload + if err := json.Unmarshal(data, &payload); err != nil { + t.Fatalf("decode mirror %q: %v", data, err) + } + return payload +} + +func TestRedisMirrorConvergesConcurrentIndependentDimensions(t *testing.T) { + ctx := context.Background() + rdb := realRedis(t) + key, redisMirrorKey := uniqueRedisRevocation(t) + t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() }) + base := time.Unix(time.Now().Add(time.Hour).Unix(), 123456789).UTC() + longer := Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "older-cutoff", + RevokedAt: base, ExpiresAt: base.Add(2 * time.Hour), + } + later := Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "later-cutoff", + RevokedAt: base.Add(time.Nanosecond), ExpiresAt: base.Add(time.Hour), + } + + stores := []*Store{New(Options{Redis: rdb}), New(Options{Redis: rdb})} + revocations := []Revocation{longer, later} + errs := make(chan error, len(stores)) + var wg sync.WaitGroup + for i := range stores { + wg.Add(1) + go func(i int) { + defer wg.Done() + errs <- stores[i].mirrorToRedis(ctx, revocations[i]) + }(i) + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + + got := readMirror(t, rdb, redisMirrorKey) + if !got.RevokedAt.Equal(later.RevokedAt) || got.Reason != later.Reason { + t.Fatalf("cutoff=%v reason=%q, want %v %q", got.RevokedAt, got.Reason, later.RevokedAt, later.Reason) + } + if !got.ExpiresAt.Equal(longer.ExpiresAt) { + t.Fatalf("expiry=%v, want %v", got.ExpiresAt, longer.ExpiresAt) + } +} + +func TestRedisMirrorPermanentKillIsNeverDowngraded(t *testing.T) { + ctx := context.Background() + rdb := realRedis(t) + key, redisMirrorKey := uniqueRedisRevocation(t) + t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() }) + base := time.Now().UTC() + store := New(Options{Redis: rdb}) + if err := store.mirrorToRedis(ctx, Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "permanent", + RevokedAt: base, + }); err != nil { + t.Fatal(err) + } + bounded := Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "new-cutoff", + RevokedAt: base.Add(time.Nanosecond), ExpiresAt: base.Add(time.Hour), + } + if err := store.mirrorToRedis(ctx, bounded); err != nil { + t.Fatal(err) + } + + got := readMirror(t, rdb, redisMirrorKey) + if !got.ExpiresAt.IsZero() { + t.Fatalf("expiry=%v, want permanent", got.ExpiresAt) + } + if !got.RevokedAt.Equal(bounded.RevokedAt) || got.Reason != bounded.Reason { + t.Fatalf("cutoff=%v reason=%q, want %v %q", got.RevokedAt, got.Reason, bounded.RevokedAt, bounded.Reason) + } +} + +func TestRedisMirrorLapsedIncomingDoesNotDeleteStrongerKill(t *testing.T) { + ctx := context.Background() + rdb := realRedis(t) + key, redisMirrorKey := uniqueRedisRevocation(t) + t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() }) + now := time.Now().UTC() + store := New(Options{Redis: rdb}) + live := Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "live", + RevokedAt: now.Add(-time.Minute), ExpiresAt: now.Add(time.Hour), + } + if err := store.mirrorToRedis(ctx, live); err != nil { + t.Fatal(err) + } + if err := store.mirrorToRedis(ctx, Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "lapsed-new-cutoff", + RevokedAt: now, ExpiresAt: now.Add(-time.Second), + }); err != nil { + t.Fatal(err) + } + + got := readMirror(t, rdb, redisMirrorKey) + if !got.ExpiresAt.Equal(live.ExpiresAt) || got.Reason != "lapsed-new-cutoff" { + t.Fatalf("merged mirror=%+v, want live expiry and newer cutoff reason", got) + } +} + +func TestRedisMirrorOverwritesLegacyPayload(t *testing.T) { + ctx := context.Background() + rdb := realRedis(t) + key, redisMirrorKey := uniqueRedisRevocation(t) + t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() }) + legacy := Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "legacy", + RevokedAt: time.Now().Add(-time.Hour), ExpiresAt: time.Now().Add(time.Hour), + } + legacyData, err := json.Marshal(legacy) + if err != nil { + t.Fatal(err) + } + if err := rdb.Set(ctx, redisMirrorKey, legacyData, time.Hour).Err(); err != nil { + t.Fatal(err) + } + incoming := Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "replacement", + RevokedAt: time.Now().UTC(), ExpiresAt: time.Now().Add(2 * time.Hour).UTC(), + } + if err := New(Options{Redis: rdb}).mirrorToRedis(ctx, incoming); err != nil { + t.Fatal(err) + } + + got := readMirror(t, rdb, redisMirrorKey) + if !got.RevokedAt.Equal(incoming.RevokedAt) || !got.ExpiresAt.Equal(incoming.ExpiresAt) || + got.RevokedAtSec == 0 || got.ExpiresAtSec == 0 { + t.Fatalf("mirror=%+v, want numeric replacement payload for %+v", got, incoming) + } +} + +func TestRedisMirrorEqualNanosecondCutoffKeepsExistingReason(t *testing.T) { + ctx := context.Background() + rdb := realRedis(t) + key, redisMirrorKey := uniqueRedisRevocation(t) + t.Cleanup(func() { _ = rdb.Del(ctx, redisMirrorKey).Err() }) + cutoff := time.Unix(time.Now().Unix(), 123456789).UTC() + first := Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "original-reason", + RevokedAt: cutoff, ExpiresAt: cutoff.Add(time.Hour), + } + second := Revocation{ + Kind: key.Kind, ID: key.ID, Reason: "second", + RevokedAt: cutoff, ExpiresAt: cutoff.Add(2 * time.Hour), + } + store := New(Options{Redis: rdb}) + if err := store.mirrorToRedis(ctx, first); err != nil { + t.Fatal(err) + } + if err := store.mirrorToRedis(ctx, second); err != nil { + t.Fatal(err) + } + + got := readMirror(t, rdb, redisMirrorKey) + if !got.RevokedAt.Equal(cutoff) || got.Reason != first.Reason { + t.Fatalf("cutoff=%v reason=%q, want stable %v %q", got.RevokedAt, got.Reason, cutoff, first.Reason) + } + if !got.ExpiresAt.Equal(second.ExpiresAt) { + t.Fatalf("expiry=%v, want independently merged %v", got.ExpiresAt, second.ExpiresAt) + } +} diff --git a/internal/streamrevoke/store_test.go b/internal/streamrevoke/store_test.go index 46ae4d51..6a767569 100644 --- a/internal/streamrevoke/store_test.go +++ b/internal/streamrevoke/store_test.go @@ -181,6 +181,9 @@ func (h *recordingRedisHook) DialHook(next redis.DialHook) redis.DialHook { func (h *recordingRedisHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook { return func(ctx context.Context, cmd redis.Cmder) error { + if cmd.Name() == "eval" || cmd.Name() == "evalsha" { + return errors.New("lua disabled") + } if cmd.Name() == "set" { select { case h.set <- struct{}{}: @@ -196,6 +199,33 @@ func (h *recordingRedisHook) ProcessPipelineHook(next redis.ProcessPipelineHook) return next } +func TestMirrorFallsBackToPlainSetWhenLuaFails(t *testing.T) { + rdb := redis.NewClient(&redis.Options{Addr: "unused.invalid:6379"}) + t.Cleanup(func() { _ = rdb.Close() }) + set := make(chan struct{}, 2) + rdb.AddHook(&recordingRedisHook{set: set}) + store := New(Options{Redis: rdb}) + var logs bytes.Buffer + previousLogger := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil))) + t.Cleanup(func() { slog.SetDefault(previousLogger) }) + revocation := Revocation{ + Kind: KindSession, ID: "fallback", RevokedAt: time.Now(), ExpiresAt: time.Now().Add(time.Hour), + } + for range 2 { + if err := store.mirrorToRedis(context.Background(), revocation); err != nil { + t.Fatal(err) + } + } + if got := len(set); got != 2 { + t.Fatalf("plain SET fallback attempts = %d, want 2", got) + } + const warning = "streamrevoke atomic Redis merge unavailable; falling back to plain mirror writes" + if got := strings.Count(logs.String(), warning); got != 1 { + t.Fatalf("fallback warnings = %d, want 1; logs: %s", got, logs.String()) + } +} + func (f *fakeDurable) UpsertTombstone(_ context.Context, tombstone Tombstone) error { f.mu.Lock() defer f.mu.Unlock()