fix(events): guard the generation counter the way the epoch key beside it is guarded (BUG-2740)

Every branch of publishScript that rotates the id space INCRs
pad:event_epoch_gen, and none checked the key first. The epoch key next to it
does, and its comment says exactly why — but the generation key never got the
same treatment.

An INCR against a corrupted key ABORTS THE SCRIPT, after the sequence INCR has
already landed. Redis is atomic against interleaving but does not roll back a
script's earlier writes, so the failure burns an id (a hole to every
receiver), repeats on the next publish, and never self-heals: the branch that
would rotate the generation is the branch that cannot run.

FOUR ABORT MODES, not the two the filing named. Measured against the pinned
miniredis rather than assumed:

  list                          -> WRONGTYPE
  hash                          -> WRONGTYPE
  string 'abc'                  -> ERR value is not an integer or out of range
  string '9223372036854775807'  -> ERR increment or decrement would overflow

A TYPE check alone — the fix the filing describes — would have covered half of
them. The value is validated on the same terms the epoch guard uses: a
positive integer of at most 18 digits, which is what the receiver's
strconv.ParseInt can read back.

RESTART AT WALL-CLOCK SECONDS, per Dave's day-49 ruling, read from the item's
trail rather than from the dispatch message. Generations only have to be
orderable among themselves, and the corrupted key is the only witness to the
previous one, so nothing can be derived from it. Restarting at 1 would put the
new generation BELOW ones receivers have already adopted, which BUG-2736's
design reads as a regression.

The seed is passed as ARGV[3] rather than read with redis.call('TIME'), and
not for a replication reason — TIME does work here, measured. It keeps the
script deterministic and lets a test inject a fixed seed and assert the EXACT
repaired value, which is what tells 'repaired' apart from 'repaired to the
wrong thing'. Clock skew between publishers is harmless: the script is atomic,
so the first to reach a corrupted key repairs it and the rest simply
increment; no two seeds are ever compared.

Repair is SET, not DEL — SET replaces a key of any type (measured), and
BUG-2736's mutation matrix already established here that a DEL is removable
without any test noticing.

Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X

internal/events/redis_bus.go

@@ -109,6 +109,34 @@ end
 return id
 `)
 
+// generationRestartSeed is the value publishScript restarts the generation
+// counter at when it finds that key corrupted (BUG-2740, ARGV[3]).
+//
+// Wall-clock SECONDS, per Dave's day-49 ruling: generations only have to be
+// orderable among themselves, and a corrupted key cannot testify to what the
+// previous generation was, so there is nothing to derive a safe increment
+// from. Seconds are strictly above any plausible increment-from-1 history
+// with no coordination and no stored state to trust.
+//
+// PASSED IN rather than read with redis.call('TIME') inside the script, for
+// two reasons and not for a replication one: the script stays deterministic,
+// and a test can inject a fixed seed and assert the exact restart value —
+// which is what lets the mutation matrix tell "repaired" apart from
+// "repaired to the wrong thing".
+//
+// Clock skew between publishers is harmless here. The script is atomic, so
+// the first publisher to reach a corrupted key repairs it and every other
+// publisher then finds a usable counter and simply increments; no two seeds
+// are ever compared with each other.
+//
+// A seam rather than a direct call so tests can pin it.
+func (b *RedisBus) generationRestartSeed() string {
+	if b.nowUnix != nil {
+		return strconv.FormatInt(b.nowUnix(), 10)
+	}
+	return strconv.FormatInt(time.Now().Unix(), 10)
+}
+
 // publishScript assigns the ID and publishes in ONE atomic Redis call. It is
 // PHASE 2 ONLY: an instance that has not been flipped still publishes through
 // the two-call path in Publish, because the bare wire form carries the ID
@@ -181,6 +209,64 @@ return id
 // — and the client is not told. The token is as durable as Redis replication
 // and no more; this narrows the window rather than closing it.
 var publishScript = redis.NewScript(`
+-- next_gen returns the next generation for the id space, REPAIRING the
+-- generation counter first if it holds something INCR cannot work with
+-- (BUG-2740).
+--
+-- Every INCR of KEYS[5] goes through here, and the reason is that an INCR
+-- against a corrupted key ABORTS THE SCRIPT — after the INCR of KEYS[1] has
+-- already advanced the sequence. Redis is atomic against interleaving but
+-- does not roll back a script's earlier writes on an error, so the failure
+-- burns an id (a hole to every receiver), repeats on the next publish, and
+-- never self-heals, because the branch that would rotate the generation is
+-- the branch that cannot run.
+--
+-- FOUR WAYS IT ABORTS, all measured against the pinned miniredis rather than
+-- assumed, because the filing named only the first two:
+--
+--   list  -> WRONGTYPE
+--   hash  -> WRONGTYPE
+--   string 'abc'                 -> ERR value is not an integer or out of range
+--   string '9223372036854775807' -> ERR increment or decrement would overflow
+--
+-- So a TYPE check alone would have covered half of them. The value has to be
+-- validated too, on exactly the terms the epoch key's guard uses next to it:
+-- a positive integer of at most 18 digits, which is what the RECEIVER's
+-- strconv.ParseInt can read back.
+--
+-- REPAIR IS SET, NOT DEL: SET replaces a key of any type (measured), and
+-- BUG-2736's mutation matrix already established here that a DEL is
+-- removable without any test noticing.
+local function next_gen()
+  local usable = false
+  local t = redis.call('TYPE', KEYS[5])['ok']
+  if t == 'none' then
+    usable = true
+  elseif t == 'string' then
+    local v = redis.call('GET', KEYS[5])
+    if string.match(v, '^[1-9][0-9]*$') and #v <= 18 then
+      usable = true
+    end
+  end
+  if not usable then
+    -- RESTARTED AT WALL-CLOCK SECONDS, not at 1 (Dave's ruling, day-49).
+    -- Generations only have to be orderable among THEMSELVES, and the
+    -- corrupted key is the only witness to what the previous one was, so it
+    -- cannot testify. A wall-clock seed is strictly above any plausible
+    -- increment-from-1 history with no coordination and nothing stored to
+    -- trust. Restarting at 1 would make the new generation LOWER than ones
+    -- receivers have already adopted, which BUG-2736's design reads as a
+    -- regression.
+    --
+    -- The seed arrives as ARGV[3] rather than from redis.call('TIME') so the
+    -- script stays deterministic and the exact restart value is assertable in
+    -- a test. (TIME does work here — measured — but nothing needs it to.)
+    redis.call('SET', KEYS[5], ARGV[3])
+    return ARGV[3]
+  end
+  return tostring(redis.call('INCR', KEYS[5]))
+end
+
 if redis.call('EXISTS', KEYS[4]) == 1 then
   return 0
 end
@@ -193,14 +279,14 @@ if id == 1 then
   -- nothing -- and the numeric check misses it too whenever a receiver's
   -- high-water mark is low enough that the restarted counter climbs past it
   -- before that receiver sees anything.
-  local g = redis.call('INCR', KEYS[5])
-  redis.call('SET', KEYS[3], tostring(g))
+  local g = next_gen()
+  redis.call('SET', KEYS[3], g)
 elseif redis.call('EXISTS', KEYS[3]) == 0 then
   -- No epoch yet for a sequence already in flight: the installation's first
   -- flipped publish, or a phase-1 instance cleared a stale one. Mint the next
   -- generation. Inside the script, so two publishers cannot both mint.
-  local g = redis.call('INCR', KEYS[5])
-  redis.call('SET', KEYS[3], tostring(g))
+  local g = next_gen()
+  redis.call('SET', KEYS[3], g)
 end
 local epoch = false
 if redis.call('TYPE', KEYS[3])['ok'] == 'string' then
@@ -228,9 +314,9 @@ if not epoch or not string.match(epoch, '^[1-9][0-9]*$') or #epoch > 18 then
   -- -- the same total, silent, unrecoverable drop by a different route. Any
   -- 18-digit number fits in an int64, and a generation counts installations'
   -- id-space resets, so 18 digits is not a bound anything real approaches.
-  local g = redis.call('INCR', KEYS[5])
-  redis.call('SET', KEYS[3], tostring(g))
-  epoch = tostring(g)
+  local g = next_gen()
+  redis.call('SET', KEYS[3], g)
+  epoch = g
 end
 redis.call('PUBLISH', KEYS[2], epoch .. '|' .. id .. '|' .. ARGV[1])
 redis.call('SET', KEYS[4], '1', 'EX', ARGV[2])
@@ -246,6 +332,12 @@ type RedisBus struct {
 
 	client *redis.Client
 
+	// nowUnix overrides the wall clock behind generationRestartSeed. Nil in
+	// every real construction; set only by tests, so the exact value a
+	// corrupted generation counter is repaired to can be asserted rather than
+	// bounded (BUG-2740).
+	nowUnix func() int64
+
 	// keys builds this installation's Redis names (BUG-2724). The zero
 	// value is the historical un-namespaced keyspace, so a bus constructed
 	// without one behaves exactly as it always did.
@@ -617,7 +709,7 @@ func (b *RedisBus) publishWithEpoch(channel string, event Event) {
 			b.keys.Name(redisSeqSuffix), channel, b.keys.Name(redisEpochSuffix),
 			dedupeKey, b.keys.Name(redisEpochGenSuffix),
 		},
-		string(data), redisDedupeTTLSeconds).Err(); err != nil {
+		string(data), redisDedupeTTLSeconds, b.generationRestartSeed()).Err(); err != nil {
 		// NO LOCAL-COUNTER FALLBACK, for the same reason the phase-1 path has
 		// none (BUG-2731): an ID minted locally belongs to a different space,
 		// which every receiving instance reads as a counter reset, and this

internal/events/redis_generation_guard_test.go

@@ -0,0 +1,191 @@
+package events
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"github.com/PerpetualSoftware/pad/internal/redisns"
+	"github.com/alicebob/miniredis/v2"
+	"github.com/redis/go-redis/v9"
+)
+
+// BUG-2740 — the generation counter had no guard, while the epoch key beside
+// it did.
+//
+// Every branch that rotates the id space INCRs pad:event_epoch_gen, and an
+// INCR against a corrupted key aborts the script AFTER the sequence INCR has
+// already landed. Redis does not roll back a script's earlier writes, so the
+// failure burns an id (a hole to every receiver), repeats on the next publish,
+// and never self-heals — the branch that would rotate the generation is the
+// branch that cannot run.
+//
+// The filing named the two WRONGTYPE cases. A probe against the pinned
+// miniredis found four, and the other two are STRING values that pass any type
+// check: a non-numeric one, and one that overflows int64 on increment. Each
+// gets its own case here for that reason — a guard that checks only TYPE
+// passes half this table.
+
+// seedFn pins the restart seed so the repaired value can be asserted exactly
+// rather than bounded, which is what lets a mutation that repairs to the WRONG
+// value be told apart from one that repairs correctly.
+const fixedSeed int64 = 1700000000
+
+func newSeededFlippedBus(t *testing.T) (*RedisBus, *miniredis.Miniredis) {
+	t.Helper()
+	mr := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { _ = client.Close() })
+	b := NewRedisBusWithKeys(client, redisns.Default, true)
+	b.nowUnix = func() int64 { return fixedSeed }
+	t.Cleanup(b.Close)
+	return b, mr
+}
+
+func TestACorruptedGenerationCounterIsRepairedRatherThanFatal(t *testing.T) {
+	genKey := redisns.Default.Name(redisEpochGenSuffix)
+	seqKey := redisns.Default.Name(redisSeqSuffix)
+	epochKey := redisns.Default.Name(redisEpochSuffix)
+
+	for _, tc := range []struct {
+		name  string
+		seed  func(t *testing.T, c *redis.Client)
+		abort string
+	}{
+		{"list", func(t *testing.T, c *redis.Client) {
+			if err := c.RPush(context.Background(), genKey, "not", "a", "counter").Err(); err != nil {
+				t.Fatalf("seed: %v", err)
+			}
+		}, "WRONGTYPE"},
+		{"hash", func(t *testing.T, c *redis.Client) {
+			if err := c.HSet(context.Background(), genKey, "f", "v").Err(); err != nil {
+				t.Fatalf("seed: %v", err)
+			}
+		}, "WRONGTYPE"},
+		{"non-numeric string", func(t *testing.T, c *redis.Client) {
+			if err := c.Set(context.Background(), genKey, "abc", 0).Err(); err != nil {
+				t.Fatalf("seed: %v", err)
+			}
+		}, "value is not an integer"},
+		{"string overflowing int64", func(t *testing.T, c *redis.Client) {
+			if err := c.Set(context.Background(), genKey, "9223372036854775807", 0).Err(); err != nil {
+				t.Fatalf("seed: %v", err)
+			}
+		}, "increment would overflow"},
+		{"zero", func(t *testing.T, c *redis.Client) {
+			if err := c.Set(context.Background(), genKey, "0", 0).Err(); err != nil {
+				t.Fatalf("seed: %v", err)
+			}
+		}, "not a positive generation"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			b, mr := newSeededFlippedBus(t)
+			next := listen(t, b.client, redisns.Default.Name(redisChannelSuffix)+"ws-1")
+			ctx := context.Background()
+
+			// PUBLISH ONCE FIRST, load-bearing rather than setup — the same
+			// trap the epoch key's test records. The id == 1 branch rotates
+			// unconditionally, so on a fresh counter the corrupted key would
+			// be repaired on a path this table is not testing. One publish
+			// puts the sequence past 1 so the later publish reaches the
+			// branch under test.
+			b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
+			if _, _, err := decodePayload(next()); err != nil {
+				t.Fatalf("fixture: the first publish must succeed, got %v", err)
+			}
+
+			// Clear the epoch so the SECOND publish takes a rotating branch,
+			// and corrupt the generation counter under it.
+			if err := b.client.Del(ctx, epochKey).Err(); err != nil {
+				t.Fatalf("clear the epoch: %v", err)
+			}
+			if err := b.client.Del(ctx, genKey).Err(); err != nil {
+				t.Fatalf("clear the generation key: %v", err)
+			}
+			tc.seed(t, b.client)
+
+			seqBefore, err := b.client.Get(ctx, seqKey).Int64()
+			if err != nil {
+				t.Fatalf("read the sequence: %v", err)
+			}
+
+			b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "item-7"})
+
+			epoch, ev, err := decodePayload(next())
+			if err != nil {
+				t.Fatalf("the publish must survive a %s generation key (%s): %v", tc.name, tc.abort, err)
+			}
+			if ev.ItemID != "item-7" {
+				t.Fatalf("the event must still carry its body, got %+v", ev)
+			}
+
+			// THE REPAIRED VALUE, ASSERTED EXACTLY. Bounding it (">0", or
+			// "large") would pass against a repair to 1 — which is the
+			// specific wrong answer the ruling exists to rule out, because a
+			// generation BELOW ones receivers have already adopted reads as a
+			// regression rather than a rotation.
+			if epoch != fixedSeed {
+				t.Fatalf("want the generation restarted at the wall-clock seed %d, got %d", fixedSeed, epoch)
+			}
+			if got := mr.Type(genKey); got != "string" {
+				t.Fatalf("the generation key must be repaired to a string, it holds %q", got)
+			}
+			if got, err := b.client.Get(ctx, genKey).Result(); err != nil || got != strconv.FormatInt(fixedSeed, 10) {
+				t.Fatalf("the generation key must hold the seed, got %q (err %v)", got, err)
+			}
+
+			// AND THE SEQUENCE ADVANCED EXACTLY ONCE. This is the leg that
+			// pins the actual damage: an aborted script leaves the sequence
+			// INCRemented with nothing published, which every receiver reads
+			// as a hole. A repair that merely stopped the error while losing
+			// the publish would satisfy every assertion above.
+			seqAfter, err := b.client.Get(ctx, seqKey).Int64()
+			if err != nil {
+				t.Fatalf("read the sequence: %v", err)
+			}
+			if seqAfter != seqBefore+1 {
+				t.Fatalf("the sequence must advance exactly once per published event: %d -> %d", seqBefore, seqAfter)
+			}
+			if ev.ID != seqAfter {
+				t.Fatalf("the published id %d must be the sequence value %d", ev.ID, seqAfter)
+			}
+		})
+	}
+}
+
+// The control: a HEALTHY generation counter is incremented, not replaced by
+// the seed. Without this leg the table above passes against a guard that
+// repairs unconditionally — which would restart the generation on every
+// rotation and make the counter meaningless.
+func TestAHealthyGenerationCounterIsIncrementedNotReseeded(t *testing.T) {
+	b, _ := newSeededFlippedBus(t)
+	epochKey := redisns.Default.Name(redisEpochSuffix)
+	next := listen(t, b.client, redisns.Default.Name(redisChannelSuffix)+"ws-1")
+	ctx := context.Background()
+
+	b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1"})
+	first, _, err := decodePayload(next())
+	if err != nil {
+		t.Fatalf("first publish: %v", err)
+	}
+	if first != 1 {
+		t.Fatalf("a fresh installation's first generation must be 1, got %d", first)
+	}
+
+	// Force another rotation with the counter intact.
+	if err := b.client.Del(ctx, epochKey).Err(); err != nil {
+		t.Fatalf("clear the epoch: %v", err)
+	}
+	b.Publish(Event{Type: ItemUpdated, WorkspaceID: "ws-1", ItemID: "item-2"})
+	second, _, err := decodePayload(next())
+	if err != nil {
+		t.Fatalf("second publish: %v", err)
+	}
+
+	if second != 2 {
+		t.Fatalf("a healthy counter must INCREMENT: want generation 2, got %d", second)
+	}
+	if second == fixedSeed {
+		t.Fatalf("a healthy counter was reseeded to the wall-clock value %d", fixedSeed)
+	}
+}