fix(playback): share the live-stream picture across replicas

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

@@ -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)

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{})
+}

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")
+	}
+}

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.

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
+}

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)
 	}
 }

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)
+	}
+}

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

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)
+	}
+}

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()