Skip to content

Commit a03fd9a

Browse files
leecalcoteleecalcote
committed
[Server] Remove destructive IsClosed signal-channel anti-pattern
The pattern that broke the events broadcaster - guard a send with utils.IsClosed(ch), then send blocking - was copied across four GraphQL subscription signal channels: SummaryChannel, DashboardK8sResourcesChan, K8scontextChan and the (unused) ConfigurationChannel. utils.IsClosed does a non-blocking receive, so on these buffered cap-1 channels it destructively consumed the pending refetch signal and then skipped the send, silently dropping the update; the blocking send could also stall the publisher. Convert every Publish to a non-blocking coalescing send taken from a snapshot under the mutex, and fix the lifecycle where it made the send unsafe: - SummaryChannel's resolver closed its channel on ctx.Done but the helper never removed it, so closed channels accumulated in the slice and a plain non-blocking send would panic on them. Add Unsubscribe and have the resolver detach instead of close. - DashboardK8sResourcesChan never removed subscribers (unbounded growth). Add Unsubscribe and defer it in the resolver. - SummaryChannel/ConfigurationChannel were seeded with 10 nil channels via make([]chan, 10); start empty. utils.IsClosed has no callers left, so remove it: checking whether a channel is closed via a receive is inherently destructive and racy and should not exist. Signed-off-by: Lee Calcote <[email protected]>
1 parent 00cc714 commit a03fd9a

7 files changed

Lines changed: 105 additions & 45 deletions

File tree

‎server/helpers/utils/utils.go‎

Lines changed: 0 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -148,18 +148,6 @@ func ToMapStringInterface(mp interface{}) map[string]interface{} {
148148
return res
149149
}
150150

151-
func IsClosed[K any](ch chan K) bool {
152-
if ch == nil {
153-
return true
154-
}
155-
select {
156-
case <-ch:
157-
return true
158-
default:
159-
}
160-
return false
161-
}
162-
163151
const UI = "../../ui/public/static/img/meshmodels" //Relative to cmd/main.go
164152
var UISVGPaths = make([]string, 1)
165153

‎server/internal/graphql/resolver/kubernetes.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,9 @@ func (r *Resolver) subscribeClusterResources(ctx context.Context, provider model
100100
r.Config.DashboardK8sResourcesChan.SubscribeDashbordK8Resources(ch)
101101

102102
go func() {
103+
// Remove this listener when the subscription ends so PublishDashboardK8sResources
104+
// stops tracking it (the channel is never closed, so removal is enough).
105+
defer r.Config.DashboardK8sResourcesChan.UnsubscribeDashboardK8sResources(ch)
103106
r.Log.Info("Initializing Cluster Resources subscription")
104107
for {
105108
select {

‎server/internal/graphql/resolver/meshmodel.go‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,10 @@ func (r *Resolver) subscribeMeshModelSummary(ctx context.Context, provider model
3232
respChan <- meshModelSummary
3333
case <-ctx.Done():
3434
close(respChan)
35-
close(ch)
35+
// Detach the listener instead of closing ch: SummaryChannel keeps
36+
// no way to remove a closed channel, so a closed ch left in its
37+
// slice would make the next Publish panic on a send to it.
38+
r.Config.MeshModelSummaryChannel.Unsubscribe(ch)
3639
r.Log.Info("Closing MeshModelSummary subscription")
3740
return
3841
}

‎server/models/configuration_helper.go‎

Lines changed: 32 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@ package models
22

33
import (
44
"sync"
5-
6-
"github.com/meshery/meshery/server/helpers/utils"
75
)
86

97
type ConfigurationChannel struct {
@@ -14,10 +12,12 @@ type ConfigurationChannel struct {
1412
}
1513

1614
func NewConfigurationHelper() *ConfigurationChannel {
15+
// Start empty; the previous length-10 make seeded each slice with 10 nil
16+
// channels that Publish then had to skip on every call.
1717
return &ConfigurationChannel{
18-
ApplicationsChannel: make([]chan struct{}, 10),
19-
PatternsChannel: make([]chan struct{}, 10),
20-
FiltersChannel: make([]chan struct{}, 10),
18+
ApplicationsChannel: make([]chan struct{}, 0),
19+
PatternsChannel: make([]chan struct{}, 0),
20+
FiltersChannel: make([]chan struct{}, 0),
2121
}
2222
}
2323

@@ -28,9 +28,15 @@ func (c *ConfigurationChannel) SubscribeApplications(ch chan struct{}) {
2828
}
2929

3030
func (c *ConfigurationChannel) PublishApplications() {
31-
for _, ch := range c.ApplicationsChannel {
32-
if !utils.IsClosed(ch) {
33-
ch <- struct{}{}
31+
c.mx.Lock()
32+
subscribers := make([]chan struct{}, len(c.ApplicationsChannel))
33+
copy(subscribers, c.ApplicationsChannel)
34+
c.mx.Unlock()
35+
36+
for _, ch := range subscribers {
37+
select {
38+
case ch <- struct{}{}:
39+
default:
3440
}
3541
}
3642
}
@@ -42,9 +48,15 @@ func (c *ConfigurationChannel) SubscribePatterns(ch chan struct{}) {
4248
}
4349

4450
func (c *ConfigurationChannel) PublishPatterns() {
45-
for _, ch := range c.PatternsChannel {
46-
if !utils.IsClosed(ch) {
47-
ch <- struct{}{}
51+
c.mx.Lock()
52+
subscribers := make([]chan struct{}, len(c.PatternsChannel))
53+
copy(subscribers, c.PatternsChannel)
54+
c.mx.Unlock()
55+
56+
for _, ch := range subscribers {
57+
select {
58+
case ch <- struct{}{}:
59+
default:
4860
}
4961
}
5062
}
@@ -56,9 +68,15 @@ func (c *ConfigurationChannel) SubscribeFilters(ch chan struct{}) {
5668
}
5769

5870
func (c *ConfigurationChannel) PublishFilters() {
59-
for _, ch := range c.FiltersChannel {
60-
if !utils.IsClosed(ch) {
61-
ch <- struct{}{}
71+
c.mx.Lock()
72+
subscribers := make([]chan struct{}, len(c.FiltersChannel))
73+
copy(subscribers, c.FiltersChannel)
74+
c.mx.Unlock()
75+
76+
for _, ch := range subscribers {
77+
select {
78+
case ch <- struct{}{}:
79+
default:
6280
}
6381
}
6482
}

‎server/models/dashboard_k8s_resources.go‎

Lines changed: 27 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@ package models
22

33
import (
44
"sync"
5-
6-
"github.com/meshery/meshery/server/helpers/utils"
75
)
86

97
type DashboardK8sResourcesChan struct {
@@ -24,10 +22,34 @@ func (d *DashboardK8sResourcesChan) SubscribeDashbordK8Resources(ch chan struct{
2422
d.ResourcesChan = append(d.ResourcesChan, ch)
2523
}
2624

25+
// UnsubscribeDashboardK8sResources removes a listener registered with
26+
// SubscribeDashbordK8Resources. Subscribers must call it when their context is
27+
// done, otherwise ResourcesChan grows without bound as subscriptions come and go.
28+
func (d *DashboardK8sResourcesChan) UnsubscribeDashboardK8sResources(ch chan struct{}) {
29+
d.mx.Lock()
30+
defer d.mx.Unlock()
31+
32+
for i := 0; i < len(d.ResourcesChan); i++ {
33+
if d.ResourcesChan[i] == ch {
34+
d.ResourcesChan = append(d.ResourcesChan[:i], d.ResourcesChan[i+1:]...)
35+
return
36+
}
37+
}
38+
}
39+
2740
func (d *DashboardK8sResourcesChan) PublishDashboardK8sResources() {
28-
for _, ch := range d.ResourcesChan {
29-
if !utils.IsClosed(ch) {
30-
ch <- struct{}{}
41+
d.mx.Lock()
42+
subscribers := make([]chan struct{}, len(d.ResourcesChan))
43+
copy(subscribers, d.ResourcesChan)
44+
d.mx.Unlock()
45+
46+
for _, ch := range subscribers {
47+
// Non-blocking coalescing send - see the note in K8scontextChan.
48+
// PublishContext. The former utils.IsClosed guard destructively consumed
49+
// buffered signals and could block the publisher on a slow subscriber.
50+
select {
51+
case ch <- struct{}{}:
52+
default:
3153
}
3254
}
3355
}

‎server/models/k8scontext_helper.go‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@ package models
22

33
import (
44
"sync"
5-
6-
"github.com/meshery/meshery/server/helpers/utils"
75
)
86

97
type K8scontextChan struct {
@@ -43,11 +41,15 @@ func (k *K8scontextChan) PublishContext() {
4341
k.mx.Unlock()
4442

4543
for _, ch := range subscribers {
46-
if !utils.IsClosed(ch) {
47-
select {
48-
case ch <- struct{}{}:
49-
default:
50-
}
44+
// Non-blocking coalescing send. A subscriber that already has a pending
45+
// signal buffered doesn't need another; a departed subscriber is removed
46+
// via UnsubscribeContext and its channel is never closed, so this can
47+
// neither block the publisher nor panic on a closed channel. The former
48+
// utils.IsClosed guard was a destructive receive that silently consumed a
49+
// buffered signal, dropping the refetch it was meant to trigger.
50+
select {
51+
case ch <- struct{}{}:
52+
default:
5153
}
5254
}
5355
}

‎server/models/meshmodel/helper.go‎

Lines changed: 30 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@ package meshmodel
22

33
import (
44
"sync"
5-
6-
"github.com/meshery/meshery/server/helpers/utils"
75
)
86

97
type SummaryChannel struct {
@@ -12,8 +10,10 @@ type SummaryChannel struct {
1210
}
1311

1412
func NewSummaryHelper() *SummaryChannel {
13+
// Start empty. The previous length-10 make seeded the slice with 10 nil
14+
// channels that Publish then had to skip on every call.
1515
return &SummaryChannel{
16-
channel: make([]chan struct{}, 10),
16+
channel: make([]chan struct{}, 0),
1717
}
1818
}
1919

@@ -23,10 +23,34 @@ func (c *SummaryChannel) Subscribe(ch chan struct{}) {
2323
c.channel = append(c.channel, ch)
2424
}
2525

26+
// Unsubscribe removes a listener registered with Subscribe. Subscribers must
27+
// call it when their context is done: it detaches the channel so Publish stops
28+
// sending to it (and so the slice doesn't grow without bound). Removing rather
29+
// than closing keeps Publish's send panic-free.
30+
func (c *SummaryChannel) Unsubscribe(ch chan struct{}) {
31+
c.mx.Lock()
32+
defer c.mx.Unlock()
33+
for i := 0; i < len(c.channel); i++ {
34+
if c.channel[i] == ch {
35+
c.channel = append(c.channel[:i], c.channel[i+1:]...)
36+
return
37+
}
38+
}
39+
}
40+
2641
func (c *SummaryChannel) Publish() {
27-
for _, ch := range c.channel {
28-
if !utils.IsClosed(ch) {
29-
ch <- struct{}{}
42+
c.mx.Lock()
43+
subscribers := make([]chan struct{}, len(c.channel))
44+
copy(subscribers, c.channel)
45+
c.mx.Unlock()
46+
47+
for _, ch := range subscribers {
48+
// Non-blocking coalescing send. The former utils.IsClosed guard did a
49+
// destructive receive that consumed a buffered refetch signal (dropping
50+
// the update) and did not actually make the following blocking send safe.
51+
select {
52+
case ch <- struct{}{}:
53+
default:
3054
}
3155
}
3256
}

0 commit comments

Comments
 (0)