Repository navigation
Conversation
A replica applies changes as Redis broadcasts them, so a write that has completed through another process can be missing from its local copy for a moment. SetAndWait only covered the writer's own reads; a process that serves the next request after another replica's write had no way to make its reads include that write. Map.Sync reads the map's revision from Redis once and waits until the local copy has applied it, so the Get, GetValues, Keys, Len, and Map calls that follow include every write that completed before Sync was called. It writes nothing and needs no coordination with other replicas: changes are applied in revision order, so reaching a revision means every earlier write is applied. It returns immediately for a never-written map, and returns an error when the context ends, the map is stopped, or Redis cannot be read; a reconnect reloads the map and also completes a pending Sync. SetAndWait and Sync now share one revision wait (waitForRevision), and setWaiter is renamed revWaiter. Tests hold back a committed write's notification to prove Sync waits for it, check reads through a second replica, and cover a never-written map, a timeout, and Close. The rmap README explains when to use SetAndWait and Sync.
Goa v3.35.0 requires Go 1.27.3, so Pulse now does too: the go directive is 1.27.3, CI tests Go 1.27 only, and the README states the new minimum. Every other direct and indirect dependency moves to its latest release, including go-redis v9.23.0, gRPC v1.85.0, and OpenTelemetry v1.47.0.
Two bugs could leave a replicated map's local copy wrong for good, and both broke the guarantee Sync advertises. TTL expiry deleted the map's hash, revision counter included, so the next write restarted at revision 1 and every replica that had seen higher revisions ignored it. And go-redis reconnects and resubscribes on its own after a lost connection without closing the subscription channel, so changes published meanwhile were never delivered and the map never reloaded. Writes now stamp each change with a generation as well as a revision. A generation starts when a write finds the hash missing (a random id stored under "=gen", revisions restart at 1); Reset and Destroy keep it. Revisions advance by exactly one per published change. A replica applies a change only when it is the next revision of its own generation, and reloads the whole map from Redis when a change is from another generation or skips a revision, and after every go-redis resubscription (run reads ChannelWithSubscriptions). The hand-written reconnect loop is deleted. Sync now reads the whole map once and replaces the local copy when Redis holds newer data (an expired map becomes empty); it writes nothing and waits for nothing. SetAndWait is Set followed by Sync, so the waiter registry, its lock, and the revision waits are deleted. A reload that skips several changes notifies subscribers with EventReset. Package pool and package streaming wrote map hashes from their own Lua scripts by copying rmap's wire format. The format now lives once, in internal/rmapwire: NewScript prepends shared Lua helpers (publishSet, publishDel, publishClear, deleteMap), and Script.Run appends the fresh generation id, which the prelude removes before the script's own code reads ARGV. All 13 rmap, 31 pool, and 6 streaming publish sites use the helpers. A guard test fails if any Lua script in the module hand-writes the protocol or calls a helper without rmapwire.NewScript. Snapshots now drop every field containing "=", which can never be a map key, so private fields pool's scheduler keeps in the same hash no longer appear after a reload. Upgrading: the published format changes, so every process that uses a replicated map must run this version; old and new versions cannot share a map.
An empty replica with no generation adopted the first change of a new generation without asking Redis. A change from an expired generation that was still queued when Sync found the map gone looked the same, so the replica brought the expired value back and kept it. A replica now reloads for every change from a generation it does not hold; a new test delivers such a stale change after Sync and checks the value stays gone. strongerEventKind and eventPriority lost their only caller when the reconnect loop was deleted and are removed. The rmap README now states the upgrade and rollback rules: every process sharing a map, including pool and streaming, must switch versions together; downgrading Pulse requires stopping every newer process and removing each map's =gen field before older versions start.
Explain that a whole-copy reload (after missed changes, expiry, or Sync) sends one notification, EventReset when the copy changed generation or skipped several changes, and that subscribers should read the map again.
The regression test now hands the stale change to the replica's message handler directly, as the run loop would, so it asserts after the change is processed instead of after an arbitrary delay. It still fails with the removed empty-copy adoption. The downgrade instructions no longer suggest Destroy: it keeps the generation field and deletes the entries. Removing =gen with HDEL is the step that keeps the data.
raphael
changed the title
A reload hid every field whose name contained "=". pool writes map hashes directly and some of its fields are real entries with "=" in them: a ticker named "a=b" is stored as "p:a=b", and pool names may start with "=". A reload made such a ticker disappear, and NewTicker then waited for it until its context ended. A snapshot now hides only rmap's own bookkeeping fields (=rev, =kind, =gen) and keeps every other field. New tests reload a hash holding "p:a=b" and "=pool:x" and create tickers named "a=b" and in a pool named "=..."; the ticker test fails with the previous filter. TestSchedule asserted the scheduler map was physically empty, which no production code relies on: the scheduler reads only its own prefixed fields and keeps private "=transition:" fields in the same hash. It now checks that no field under the scheduler's ownership prefix remains.
NewTicker waits for its stored field before returning, so a regression that hid the field would block creation until the suite timeout. The test now creates each ticker with a five-second context, and fails with a deadline error instead.
The generation lived in a new "=gen" field of the map's hash. Sink names are unrestricted and streaming stores each sink's membership under its name in a replicated map, so once any sink had initialized the generation, creating a sink named "=gen" failed with SINKMEMBERSHIPINVALID, and a reload hid that sink's membership. The hash now keeps only the two fields Pulse has always reserved: "=rev" holds the revision and "=kind" holds the last change's kind and the generation as "<kind>:<gen>". stamp takes the change's kind, reads the generation from "=kind", and starts one when there is none. Every other field, "=gen" included, belongs to the map's writers again. Downgrading no longer needs data cleanup: older versions already hide "=kind" and treat its new value as an ordinary change, so stopping every newer process before an older one starts is enough. The README, the test helper's tombstone check, and the guard test follow. A new streaming test creates a sink and then one named "=gen"; it fails with the previous field.
refresh kept the local copy whenever the snapshot was in the same generation with an equal or lower revision. A map written by an older Pulse carries no generation, so when it expired before any new write, Sync read an empty hash with revision 0 in the same empty generation, kept the expired entries, and returned success. refresh holds the map lock while it reads, so the snapshot is never older than the copy: every change the copy holds was already in Redis. It now skips only an identical generation and revision, and applies anything else, including a lower revision, which means the data is gone; that case notifies EventReset. A new test loads a map in the old format, lets it expire for real, and checks Sync empties the copy and notifies a reset; it fails with the previous check.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
A replicated map (
rmap.Map) serves reads from each process's in-memory copy, which Redis keeps current by broadcasting every change. Two existing bugs could leave that copy wrong for good:WithTTLorWithSlidingTTLexpired, its revision counter disappeared with its data. The next write was numbered 1, and every process that had seen higher revisions ignored it and kept serving the expired values.Processes also had no way to make their reads include a write another process had just finished. For example, a request that saves settings through process A, followed by a request that reads them through process B, could see the old settings.
What changes
=kindfield) and revisions restart at 1. Revisions advance by exactly one per published change, so a process applies a change only when it is the next revision of its own generation, and reloads the whole map from Redis when a change comes from another generation or skips a revision.ChannelWithSubscriptions). The hand-written reconnect loop is deleted.Map.Sync(ctx)reads the whole map once and replaces the local copy if Redis holds newer data, so later reads include every write that finished before the call. An expired map becomes empty.SetAndWaitis nowSetfollowed bySync, which deletes the waiter registry and its lock.poolandstreamingwrote map hashes from their own Lua scripts by copying rmap's message format. That format now lives once, ininternal/rmapwire:NewScriptadds shared Lua helpers (publishSet,publishDel,publishClear,deleteMap), andRunsupplies the generation id, removing it before the script's own code reads its arguments. All 13 rmap, 31 pool, and 6 streaming publish sites use the helpers, and a guard test fails if any script writes the format by hand.=revand=kind). Every other field thatpoolorstreamingstores in a map's hash stays visible, whatever its name.Upgrading: the published change format changes, so every process that shares a map, including users of
poolandstreaming, must run this version together. Downgrading Pulse only requires stopping every newer process before an older one starts; no data cleanup is needed. The rmap README documents both.Tests commit a change in Redis without publishing it, let a map expire for real while three processes hold it, drop one process's own subscriber connection while another writes, and deliver a stale change from an expired generation. Each fails against the previous code. The full race suite passes on Go 1.27.3.