fix(concurrency): guard the three shared maps with a mutex - #196
Conversation
clientPointer, myClientPointer and killChannel are created once at startup and
passed by reference into eleven packages. All of them read and wrote those plain
Go maps from different goroutines, with no synchronisation.
Go does not tolerate that: the runtime kills the process with
fatal error: concurrent map writes
and a fatal error is not recoverable — a deferred recover() does not catch it.
Measured in production: the process died inside StartClient while five instances
were connecting at once. The orchestrator restarted it within seconds, but every
number was offline in the meantime, and it repeats on any restart that brings up
more than a couple of instances together.
The three maps become safemap.Map[T], a generic wrapper over a RWMutex. It lives
in its own package because MyClient is defined in pkg/whatsmeow/service, and
putting the type there would create an import cycle with the ten packages that
receive the maps.
98 call sites changed across 12 files, with the compiler as the judge: anything
missed is a build error, not a silent bug.
Tests: with -race, a plain map running the same exercise reports DATA RACE and
fails, while safemap.Map passes. Both halves verified.
Reviewer's GuideReplaces the three unsynchronized process-wide maps with pointer-shared generic maps guarded by RWMutexes, then mechanically updates all consumers to use synchronized accessors while preserving map semantics and safe snapshot-based iteration. Dedicated race-enabled tests validate concurrent access and zero-value reads. Sequence diagram for concurrent client startupsequenceDiagram
participant InstanceService
participant WhatsmeowService
participant ClientMap as safemap.Map[Client]
participant MyClientMap as safemap.Map[MyClient]
par instance A startup
InstanceService->>WhatsmeowService: StartClient(cd)
WhatsmeowService->>ClientMap: Set(instanceId, client)
ClientMap-->>WhatsmeowService: write completes
WhatsmeowService->>MyClientMap: Set(instanceId, mycli)
MyClientMap-->>WhatsmeowService: write completes
and instance B startup
InstanceService->>WhatsmeowService: StartClient(cd)
WhatsmeowService->>ClientMap: Set(instanceId, client)
ClientMap-->>WhatsmeowService: RWMutex serializes write
WhatsmeowService->>MyClientMap: Set(instanceId, mycli)
MyClientMap-->>WhatsmeowService: RWMutex serializes write
end
Flow diagram for safe map reads and iterationflowchart TD
Read[Consumer needs shared state] --> Lookup{Single-key access?}
Lookup -->|yes| Get[Get or Lookup]
Lookup -->|write| Set[Set or Delete]
Lookup -->|range| Snapshot[Snapshot]
Get --> ReadLock[RWMutex read lock]
Snapshot --> Copy[Copy map under read lock]
Copy --> Iterate[Range copied map]
Iterate --> Network[Perform WhatsApp or network calls without holding map lock]
Set --> WriteLock[RWMutex write lock]
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 3 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="pkg/whatsmeow/service/whatsmeow.go" line_range="312-313" />
<code_context>
- if w.clientPointer[cd.Instance.Id] != nil {
- if w.clientPointer[cd.Instance.Id].IsConnected() {
+ if w.clientPointer.Get(cd.Instance.Id) != nil {
+ if w.clientPointer.Get(cd.Instance.Id).IsConnected() {
return
}
</code_context>
<issue_to_address>
**issue (bug_risk):** The initial `Get` and later `Set` are not atomic, so two concurrent `StartClient` calls for the same instance both pass the connected check and create clients; the later `Set` overwrites the first pointer while the first client's connection and event handler remain active and unmanaged.
**Triggers:** When the same instance is started or reconnected concurrently.
**Suggested fix:** Serialize start/reconnect per instance, or add an atomic compare-and-set/claim operation to the synchronized map.
</issue_to_address>
### Comment 2
<location path="pkg/whatsmeow/service/whatsmeow.go" line_range="2386" />
<code_context>
}
- w.killChannel[instance.Id] = make(chan bool)
+ w.killChannel.Set(instance.Id, make(chan bool))
clientData := &ClientData{
</code_context>
<issue_to_address>
**issue (bug_risk):** `StartInstance` replaces the existing kill channel without coordinating with the existing `StartClient` goroutine. Concurrent starts can leave multiple clients listening to one replacement channel while the original channel is orphaned, so a single kill signal stops only one client and leaves another running.
**Triggers:** When the same instance is started more than once before the previous client exits.
**Suggested fix:** Make channel installation and client startup a per-instance atomic lifecycle operation, or reject a start while an existing client is active.
</issue_to_address>
### Comment 3
<location path="pkg/safemap/safemap_test.go" line_range="22" />
<code_context>
+//
+// go test -race -run TestPlainMapRaces ./pkg/safemap/
+func TestPlainMapRaces(t *testing.T) {
+ t.Skip("calibration: run by hand — see the comment above")
+
+ m := map[string]int{}
</code_context>
<issue_to_address>
**issue (testing):** The calibration test unconditionally calls `t.Skip`, so the documented `go test -race -run TestPlainMapRaces ./pkg/safemap/` command never executes the plain-map race and cannot verify that the detector catches the intended failure.
**Triggers:** When relying on the documented calibration command to validate race-detector coverage.
**Suggested fix:** Gate the skip behind an explicit environment variable or separate the calibration test into a manually enabled test target.
</issue_to_address>Sourcery assessment
Needs a human reviewer. 3 findings to address first, and this changes shared runtime state across many services, and individual map operations are locked but sequences such as Get-then-send or Get-then-close are not atomic; a concurrent logout or cleanup could still panic on a closed channel or act on a stale client. Reverting removes the new race behavior, but any crash or service outage that already occurred would not be undone.
Blocking findings: pkg/whatsmeow/service/whatsmeow.go:313, pkg/whatsmeow/service/whatsmeow.go:2386, pkg/safemap/safemap_test.go:22
| if w.clientPointer.Get(cd.Instance.Id) != nil { | ||
| if w.clientPointer.Get(cd.Instance.Id).IsConnected() { |
There was a problem hiding this comment.
issue (bug_risk): The initial Get and later Set are not atomic, so two concurrent StartClient calls for the same instance both pass the connected check and create clients; the later Set overwrites the first pointer while the first client's connection and event handler remain active and unmanaged.
Triggers: When the same instance is started or reconnected concurrently.
Suggested fix: Serialize start/reconnect per instance, or add an atomic compare-and-set/claim operation to the synchronized map.
| } | ||
|
|
||
| w.killChannel[instance.Id] = make(chan bool) | ||
| w.killChannel.Set(instance.Id, make(chan bool)) |
There was a problem hiding this comment.
issue (bug_risk): StartInstance replaces the existing kill channel without coordinating with the existing StartClient goroutine. Concurrent starts can leave multiple clients listening to one replacement channel while the original channel is orphaned, so a single kill signal stops only one client and leaves another running.
Triggers: When the same instance is started more than once before the previous client exits.
Suggested fix: Make channel installation and client startup a per-instance atomic lifecycle operation, or reject a start while an existing client is active.
| // | ||
| // go test -race -run TestPlainMapRaces ./pkg/safemap/ | ||
| func TestPlainMapRaces(t *testing.T) { | ||
| t.Skip("calibration: run by hand — see the comment above") |
There was a problem hiding this comment.
issue (testing): The calibration test unconditionally calls t.Skip, so the documented go test -race -run TestPlainMapRaces ./pkg/safemap/ command never executes the plain-map race and cannot verify that the detector catches the intended failure.
Triggers: When relying on the documented calibration command to validate race-detector coverage.
Suggested fix: Gate the skip behind an explicit environment variable or separate the calibration test into a manually enabled test target.
2 participants
The problem
clientPointer,myClientPointerandkillChannelare plain Go maps, created once at startup and passed by reference into eleven packages. All of them read and write those maps from different goroutines, with no synchronisation.Go does not tolerate that. Concurrent writes do not corrupt silently — the runtime kills the process:
Line 550 is
w.clientPointer[cd.Instance.Id] = client.StartInstancelaunches one goroutine per instance, so the window opens whenever more than a couple of instances connect at the same time — most commonly at startup, sinceCONNECT_ON_STARTUPbrings them all up together.A
fatal erroris not recoverable. A deferredrecover()does not catch it, so no amount of defensive code around the call site helps.Measured
Our process died exactly this way while five instances were connecting. The orchestrator restarted it within seconds, but every number was offline in the meantime, and it recurs on any restart of that size.
The fix
The three maps become
safemap.Map[T], a small generic wrapper over async.RWMutex:m[k]m.Get(k)v, ok := m[k]m.Lookup(k)m[k] = vm.Set(k, v)delete(m, k)m.Delete(k)len(m)m.Len()range mrange m.Snapshot()Getreturns the zero value for an absent key, exactly likem[k], so swapping the call sites changes no behaviour. There is a test pinning that.Why a separate package.
MyClientis defined inpkg/whatsmeow/service. Putting the type there would force the other ten packages to import it and create an import cycle.pkg/safemaphas no internal dependencies.Why
Snapshotfor ranges. Ranging from inside would hold the lock for the whole loop, and these loop bodies talk to WhatsApp — the lock would be held across network calls.Scope
98 call sites across 12 files, plus the new package. The changes are mechanical and the compiler is the judge: anything missed is a build error, not a silent bug.
Testing
go build ./...againstmain, clean.go test -race ./pkg/safemap/passes.DATA RACEand fails — proof that the detector is not blind before trusting the green result.Summary by Sourcery
Protect shared runtime state from concurrent goroutine access by replacing the process-wide maps with synchronized generic maps.
Bug Fixes:
Enhancements:
Tests: