Skip to content

Commit c693ec3

Browse files
authored
[FIXED] Deadlock when accessing subscriptions map on consumer (#1671)
This fixes an issue where a deadlock could occur when calling `Stop()` or `Drain()` on `ConsumeContext` or `MessagesContext` and then calling `Consume` or `Messages` immediately. Switched to using a type-safe implementation of `sync.Map` for subscriptions map instead of locking the whole consumer state. Additionally, changed the type of atomic flags from `uint32` to `atomic.UInt32` to avoid accidental non-atomic reads/writes. Signed-off-by: Piotr Piotrowski <piotr@synadia.com> --------- Signed-off-by: Piotr Piotrowski <piotr@synadia.com>
1 parent 94fa0cb commit c693ec3

8 files changed

Lines changed: 328 additions & 138 deletions

File tree

go_test.mod

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,25 @@
11
module github.com/nats-io/nats.go
22

3-
go 1.19
3+
go 1.21
4+
5+
toolchain go1.22.5
46

57
require (
68
github.com/golang/protobuf v1.4.2
7-
github.com/klauspost/compress v1.17.8
9+
github.com/klauspost/compress v1.17.9
810
github.com/nats-io/jwt v1.2.2
9-
github.com/nats-io/nats-server/v2 v2.10.16
11+
github.com/nats-io/nats-server/v2 v2.10.17
1012
github.com/nats-io/nkeys v0.4.7
1113
github.com/nats-io/nuid v1.0.1
1214
go.uber.org/goleak v1.3.0
13-
golang.org/x/text v0.15.0
15+
golang.org/x/text v0.16.0
1416
google.golang.org/protobuf v1.23.0
1517
)
1618

1719
require (
1820
github.com/minio/highwayhash v1.0.2 // indirect
1921
github.com/nats-io/jwt/v2 v2.5.7 // indirect
20-
golang.org/x/crypto v0.23.0 // indirect
21-
golang.org/x/sys v0.20.0 // indirect
22+
golang.org/x/crypto v0.24.0 // indirect
23+
golang.org/x/sys v0.21.0 // indirect
2224
golang.org/x/time v0.5.0 // indirect
2325
)

go_test.sum

Lines changed: 14 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
2+
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
23
github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
34
github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
45
github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
@@ -10,38 +11,40 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
1011
github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
1112
github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
1213
github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
13-
github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=
14-
github.com/klauspost/compress v1.17.8/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
14+
github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
15+
github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
1516
github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
1617
github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
1718
github.com/nats-io/jwt v1.2.2 h1:w3GMTO969dFg+UOKTmmyuu7IGdusK+7Ytlt//OYH/uU=
1819
github.com/nats-io/jwt v1.2.2/go.mod h1:/xX356yQA6LuXI9xWW7mZNpxgF2mBmGecH+Fj34sP5Q=
1920
github.com/nats-io/jwt/v2 v2.5.7 h1:j5lH1fUXCnJnY8SsQeB/a/z9Azgu2bYIDvtPVNdxe2c=
2021
github.com/nats-io/jwt/v2 v2.5.7/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
21-
github.com/nats-io/nats-server/v2 v2.10.16 h1:2jXaiydp5oB/nAx/Ytf9fdCi9QN6ItIc9eehX8kwVV0=
22-
github.com/nats-io/nats-server/v2 v2.10.16/go.mod h1:Pksi38H2+6xLe1vQx0/EA4bzetM0NqyIHcIbmgXSkIU=
22+
github.com/nats-io/nats-server/v2 v2.10.17 h1:PTVObNBD3TZSNUDgzFb1qQsQX4mOgFmOuG9vhT+KBUY=
23+
github.com/nats-io/nats-server/v2 v2.10.17/go.mod h1:5OUyc4zg42s/p2i92zbbqXvUNsbF0ivdTLKshVMn2YQ=
2324
github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
2425
github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
2526
github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
2627
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
2728
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
2829
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
30+
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
2931
github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
32+
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
3033
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
3134
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
3235
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
3336
golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
34-
golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
35-
golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
37+
golang.org/x/crypto v0.24.0 h1:mnl8DM0o513X8fdIkmyFE/5hTYxbwYOjDS/+rK6qpRI=
38+
golang.org/x/crypto v0.24.0/go.mod h1:Z1PMYSOR5nyMcyAVAIQSKCDwalqy85Aqn1x3Ws4L5DM=
3639
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
3740
golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
3841
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
3942
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
40-
golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
41-
golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
43+
golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws=
44+
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
4245
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
43-
golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk=
44-
golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
46+
golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
47+
golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
4548
golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
4649
golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
4750
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
@@ -54,3 +57,4 @@ google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzi
5457
google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
5558
google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
5659
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
60+
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/syncx/map.go

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
// Copyright 2024 The NATS Authors
2+
// Licensed under the Apache License, Version 2.0 (the "License");
3+
// you may not use this file except in compliance with the License.
4+
// You may obtain a copy of the License at
5+
//
6+
// http://www.apache.org/licenses/LICENSE-2.0
7+
//
8+
// Unless required by applicable law or agreed to in writing, software
9+
// distributed under the License is distributed on an "AS IS" BASIS,
10+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11+
// See the License for the specific language governing permissions and
12+
// limitations under the License.
13+
14+
package syncx
15+
16+
import "sync"
17+
18+
// Map is a type-safe wrapper around sync.Map.
19+
// It is safe for concurrent use.
20+
// The zero value of Map is an empty map ready to use.
21+
type Map[K comparable, V any] struct {
22+
m sync.Map
23+
}
24+
25+
func (m *Map[K, V]) Load(key K) (V, bool) {
26+
v, ok := m.m.Load(key)
27+
if !ok {
28+
var empty V
29+
return empty, false
30+
}
31+
return v.(V), true
32+
}
33+
34+
func (m *Map[K, V]) Store(key K, value V) {
35+
m.m.Store(key, value)
36+
}
37+
38+
func (m *Map[K, V]) Delete(key K) {
39+
m.m.Delete(key)
40+
}
41+
42+
func (m *Map[K, V]) Range(f func(key K, value V) bool) {
43+
m.m.Range(func(key, value any) bool {
44+
return f(key.(K), value.(V))
45+
})
46+
}
47+
48+
func (m *Map[K, V]) LoadOrStore(key K, value V) (V, bool) {
49+
v, loaded := m.m.LoadOrStore(key, value)
50+
return v.(V), loaded
51+
}
52+
53+
func (m *Map[K, V]) LoadAndDelete(key K) (V, bool) {
54+
v, ok := m.m.LoadAndDelete(key)
55+
if !ok {
56+
var empty V
57+
return empty, false
58+
}
59+
return v.(V), true
60+
}
61+
62+
func (m *Map[K, V]) CompareAndSwap(key K, old, new V) bool {
63+
return m.m.CompareAndSwap(key, old, new)
64+
}
65+
66+
func (m *Map[K, V]) CompareAndDelete(key K, value V) bool {
67+
return m.m.CompareAndDelete(key, value)
68+
}
69+
70+
func (m *Map[K, V]) Swap(key K, value V) (V, bool) {
71+
previous, loaded := m.m.Swap(key, value)
72+
return previous.(V), loaded
73+
}

internal/syncx/map_test.go

Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,152 @@
1+
// Copyright 2024 The NATS Authors
2+
// Licensed under the Apache License, Version 2.0 (the "License");
3+
// you may not use this file except in compliance with the License.
4+
// You may obtain a copy of the License at
5+
//
6+
// http://www.apache.org/licenses/LICENSE-2.0
7+
//
8+
// Unless required by applicable law or agreed to in writing, software
9+
// distributed under the License is distributed on an "AS IS" BASIS,
10+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11+
// See the License for the specific language governing permissions and
12+
// limitations under the License.
13+
14+
package syncx
15+
16+
import (
17+
"testing"
18+
)
19+
20+
func TestMapLoad(t *testing.T) {
21+
var m Map[int, string]
22+
m.Store(1, "one")
23+
24+
v, ok := m.Load(1)
25+
if !ok || v != "one" {
26+
t.Errorf("Load(1) = %v, %v; want 'one', true", v, ok)
27+
}
28+
29+
v, ok = m.Load(2)
30+
if ok || v != "" {
31+
t.Errorf("Load(2) = %v, %v; want '', false", v, ok)
32+
}
33+
}
34+
35+
func TestMapStore(t *testing.T) {
36+
var m Map[int, string]
37+
m.Store(1, "one")
38+
39+
v, ok := m.Load(1)
40+
if !ok || v != "one" {
41+
t.Errorf("Load(1) after Store(1, 'one') = %v, %v; want 'one', true", v, ok)
42+
}
43+
}
44+
45+
func TestMapDelete(t *testing.T) {
46+
var m Map[int, string]
47+
m.Store(1, "one")
48+
m.Delete(1)
49+
50+
v, ok := m.Load(1)
51+
if ok || v != "" {
52+
t.Errorf("Load(1) after Delete(1) = %v, %v; want '', false", v, ok)
53+
}
54+
}
55+
56+
func TestMapRange(t *testing.T) {
57+
var m Map[int, string]
58+
m.Store(1, "one")
59+
m.Store(2, "two")
60+
61+
var keys []int
62+
var values []string
63+
m.Range(func(key int, value string) bool {
64+
keys = append(keys, key)
65+
values = append(values, value)
66+
return true
67+
})
68+
69+
if len(keys) != 2 || len(values) != 2 {
70+
t.Errorf("Range() keys = %v, values = %v; want 2 keys and 2 values", keys, values)
71+
}
72+
}
73+
74+
func TestMapLoadOrStore(t *testing.T) {
75+
var m Map[int, string]
76+
77+
v, loaded := m.LoadOrStore(1, "one")
78+
if loaded || v != "one" {
79+
t.Errorf("LoadOrStore(1, 'one') = %v, %v; want 'one', false", v, loaded)
80+
}
81+
82+
v, loaded = m.LoadOrStore(1, "uno")
83+
if !loaded || v != "one" {
84+
t.Errorf("LoadOrStore(1, 'uno') = %v, %v; want 'one', true", v, loaded)
85+
}
86+
}
87+
88+
func TestMapLoadAndDelete(t *testing.T) {
89+
var m Map[int, string]
90+
m.Store(1, "one")
91+
92+
v, ok := m.LoadAndDelete(1)
93+
if !ok || v != "one" {
94+
t.Errorf("LoadAndDelete(1) = %v, %v; want 'one', true", v, ok)
95+
}
96+
97+
v, ok = m.Load(1)
98+
if ok || v != "" {
99+
t.Errorf("Load(1) after LoadAndDelete(1) = %v, %v; want '', false", v, ok)
100+
}
101+
102+
// Test that LoadAndDelete on a missing key returns the zero value.
103+
v, ok = m.LoadAndDelete(2)
104+
if ok || v != "" {
105+
t.Errorf("LoadAndDelete(2) = %v, %v; want '', false", v, ok)
106+
}
107+
}
108+
109+
func TestMapCompareAndSwap(t *testing.T) {
110+
var m Map[int, string]
111+
m.Store(1, "one")
112+
113+
ok := m.CompareAndSwap(1, "one", "uno")
114+
if !ok {
115+
t.Errorf("CompareAndSwap(1, 'one', 'uno') = false; want true")
116+
}
117+
118+
v, _ := m.Load(1)
119+
if v != "uno" {
120+
t.Errorf("Load(1) after CompareAndSwap = %v; want 'uno'", v)
121+
}
122+
}
123+
124+
func TestMapCompareAndDelete(t *testing.T) {
125+
var m Map[int, string]
126+
m.Store(1, "one")
127+
128+
ok := m.CompareAndDelete(1, "one")
129+
if !ok {
130+
t.Errorf("CompareAndDelete(1, 'one') = false; want true")
131+
}
132+
133+
v, _ := m.Load(1)
134+
if v != "" {
135+
t.Errorf("Load(1) after CompareAndDelete = %v; want ''", v)
136+
}
137+
}
138+
139+
func TestMapSwap(t *testing.T) {
140+
var m Map[int, string]
141+
m.Store(1, "one")
142+
143+
v, loaded := m.Swap(1, "uno")
144+
if !loaded || v != "one" {
145+
t.Errorf("Swap(1, 'uno') = %v, %v; want 'one', true", v, loaded)
146+
}
147+
148+
v, _ = m.Load(1)
149+
if v != "uno" {
150+
t.Errorf("Load(1) after Swap = %v; want 'uno'", v)
151+
}
152+
}

jetstream/consumer.go

Lines changed: 13 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"fmt"
2121
"strings"
2222

23+
"github.com/nats-io/nats.go/internal/syncx"
2324
"github.com/nats-io/nuid"
2425
)
2526

@@ -233,12 +234,12 @@ func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg Consu
233234
}
234235

235236
return &pullConsumer{
236-
jetStream: js,
237-
stream: stream,
238-
name: resp.Name,
239-
durable: cfg.Durable != "",
240-
info: resp.ConsumerInfo,
241-
subscriptions: make(map[string]*pullSubscription),
237+
jetStream: js,
238+
stream: stream,
239+
name: resp.Name,
240+
durable: cfg.Durable != "",
241+
info: resp.ConsumerInfo,
242+
subs: syncx.Map[string, *pullSubscription]{},
242243
}, nil
243244
}
244245

@@ -285,12 +286,12 @@ func getConsumer(ctx context.Context, js *jetStream, stream, name string) (Consu
285286
}
286287

287288
cons := &pullConsumer{
288-
jetStream: js,
289-
stream: stream,
290-
name: name,
291-
durable: resp.Config.Durable != "",
292-
info: resp.ConsumerInfo,
293-
subscriptions: make(map[string]*pullSubscription, 0),
289+
jetStream: js,
290+
stream: stream,
291+
name: name,
292+
durable: resp.Config.Durable != "",
293+
info: resp.ConsumerInfo,
294+
subs: syncx.Map[string, *pullSubscription]{},
294295
}
295296

296297
return cons, nil

0 commit comments

Comments
 (0)