diff --git a/internal/events/redis_bus.go b/internal/events/redis_bus.go index fc4aabee..32e3d75b 100644 --- a/internal/events/redis_bus.go +++ b/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 diff --git a/internal/events/redis_generation_guard_test.go b/internal/events/redis_generation_guard_test.go new file mode 100644 index 00000000..395ae7ac --- /dev/null +++ b/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) + } +}