From 11d0c9652ad59a8d2bdb61d9f67ee02d5aa69fd1 Mon Sep 17 00:00:00 2001 From: Lee Calcote Date: Sat, 4 Jul 2026 18:51:01 -0500 Subject: [PATCH 1/5] fix(meshsync): move exec/log-stream sessions out of the shared channelPool channelPool held both the fixed system channels (Stop/OS/ReSync) and dynamic per-request exec/log-stream session channels. Session goroutines mutated the map (add/delete) while other goroutines read it (system-channel selects, getActiveChannels) - a data race that can panic the process. Dynamic sessions now live in a dedicated mutex-guarded sessions map; channelPool is read-only after construction. The session helpers return the channel and release the lock before any channel send/receive, so the mutex is never held across a blocking channel op. Also fixes getActiveChannels, which previously ranged the whole pool and reported system-channel keys as active sessions. Addresses the channelPool race flagged on #573 (tracked in #585). The exec input-subscription teardown (the other half of #585) still needs a MeshKit broker Unsubscribe and is not covered here. Signed-off-by: marblom007 <158522975+marblom007@users.noreply.github.com> Signed-off-by: Lee Calcote --- meshsync/exec.go | 62 +++++++++++++++------------------- meshsync/logstream.go | 29 ++++++++-------- meshsync/meshsync.go | 14 ++++++-- meshsync/sessions.go | 57 +++++++++++++++++++++++++++++++ meshsync/sessions_test.go | 71 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 180 insertions(+), 53 deletions(-) create mode 100644 meshsync/sessions.go create mode 100644 meshsync/sessions_test.go diff --git a/meshsync/exec.go b/meshsync/exec.go index cb36dee0..37cd64cf 100644 --- a/meshsync/exec.go +++ b/meshsync/exec.go @@ -72,28 +72,26 @@ func (h *Handler) processExecRequest(obj interface{}, cfg config.ListenerConfig) for _, req := range reqs { id := fmt.Sprintf("exec.%s.%s.%s.%s", req.Namespace, req.Name, req.Container, req.ID) - if _, ok := h.channelPool[id]; !ok { - // Subscribing the first time - if !bool(req.Stop) { - h.channelPool[id] = channels.NewStructChannel() - h.Log.Debug("Starting session") - - err := h.Broker.Publish("active_sessions.exec", &broker.Message{ - ObjectType: broker.ActiveExecObject, - Object: h.getActiveChannels(), - }) - if err != nil { - h.Log.Error(ErrGetObject(err)) - } - go h.streamSession(id, req, cfg) - } - } else { - // Already running subscription - if bool(req.Stop) { - // TODO: once we have a unsubscribe functionality, need to publish message to active sessions subject - execCleanup(h, id) - } + if bool(req.Stop) { + // Stop request: tear down the session if one is running (no-op otherwise). + // TODO: once we have unsubscribe functionality, publish to active sessions subject. + execCleanup(h, id) + continue } + if _, created := h.addSession(id); !created { + // A session for this id is already running. + continue + } + h.Log.Debug("Starting session") + + err := h.Broker.Publish("active_sessions.exec", &broker.Message{ + ObjectType: broker.ActiveExecObject, + Object: h.getActiveChannels(), + }) + if err != nil { + h.Log.Error(ErrGetObject(err)) + } + go h.streamSession(id, req, cfg) } return nil @@ -104,9 +102,10 @@ func (h *Handler) processActiveExecRequest() error { return nil } func (h *Handler) getActiveChannels() []*string { - activeChannels := make([]*string, 0, len(h.channelPool)) - for k := range h.channelPool { - activeChannels = append(activeChannels, &k) + ids := h.activeSessionIDs() + activeChannels := make([]*string, 0, len(ids)) + for i := range ids { + activeChannels = append(activeChannels, &ids[i]) } return activeChannels @@ -164,7 +163,7 @@ func (h *Handler) streamSession(id string, req model.ExecRequest, cfg config.Lis _ = tstdin.Close() _ = stdout.Close() _ = getStdout.Close() - delete(h.channelPool, id) + h.deleteSession(id) }) } defer terminate() @@ -263,10 +262,8 @@ func (h *Handler) streamSession(id string, req model.ExecRequest, cfg config.Lis }() for { - // The session's StructChannel is asserted below; once terminate() has - // removed the pool entry that assertion would be on a nil interface and - // panic, so bail out first if the session is already gone. - sessionCh, ok := h.channelPool[id].(channels.StructChannel) + // If terminate() has already removed the session, bail out. + sessionCh, ok := h.getSession(id) if !ok { h.Log.Debugf("Session closed for: %s", id) return @@ -293,12 +290,7 @@ func (h *Handler) streamSession(id string, req model.ExecRequest, cfg config.Lis } func execCleanup(h *Handler, id string) { - ch, ok := h.channelPool[id] - if !ok { - return - } - - structChan, ok := ch.(channels.StructChannel) + structChan, ok := h.getSession(id) if !ok { return } diff --git a/meshsync/logstream.go b/meshsync/logstream.go index 2f0a03be..2d2627e6 100644 --- a/meshsync/logstream.go +++ b/meshsync/logstream.go @@ -7,7 +7,6 @@ import ( "io" "github.com/meshery/meshkit/broker" - "github.com/meshery/meshsync/internal/channels" "github.com/meshery/meshsync/internal/config" "github.com/meshery/meshsync/pkg/model" v1 "k8s.io/api/core/v1" @@ -27,17 +26,15 @@ func (h *Handler) processLogRequest(obj interface{}, cfg config.ListenerConfig) for _, req := range reqs { id := fmt.Sprintf("logs.%s.%s.%s", req.Namespace, req.Name, req.Container) - if _, ok := h.channelPool[id]; !ok { - // Subscribing the first time - if !bool(req.Stop) { - h.channelPool[id] = channels.NewStructChannel() - go h.streamLogs(id, req, cfg) - } - } else { - // Already running subscription - if bool(req.Stop) { - h.channelPool[id].(channels.StructChannel) <- struct{}{} + if bool(req.Stop) { + // Stop request: signal the running stream, if any, to close. + if ch, ok := h.getSession(id); ok { + ch <- struct{}{} } + continue + } + if _, created := h.addSession(id); created { + go h.streamLogs(id, req, cfg) } } @@ -58,14 +55,16 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene }).Stream(context.TODO()) if err != nil { h.Log.Error(ErrLogStream(err)) - delete(h.channelPool, id) + h.deleteSession(id) return } go func() { - <-h.channelPool[id].(channels.StructChannel) + if ch, ok := h.getSession(id); ok { + <-ch + } h.Log.Debugf("Closing %s", id) - delete(h.channelPool, id) + h.deleteSession(id) resp.Close() }() @@ -80,7 +79,7 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene } if err != nil { h.Log.Error(ErrCopyBuffer(err)) - delete(h.channelPool, id) + h.deleteSession(id) } message := string(buf[:numBytes]) diff --git a/meshsync/meshsync.go b/meshsync/meshsync.go index 64ab1075..ed63accd 100644 --- a/meshsync/meshsync.go +++ b/meshsync/meshsync.go @@ -1,6 +1,8 @@ package meshsync import ( + "sync" + "github.com/meshery/meshkit/broker" "github.com/meshery/meshkit/config" "github.com/meshery/meshkit/logger" @@ -22,10 +24,15 @@ type Handler struct { Log logger.Handler Broker broker.Handler - clusterID string - informer dynamicinformer.DynamicSharedInformerFactory - kubeClient *mesherykube.Client + clusterID string + informer dynamicinformer.DynamicSharedInformerFactory + kubeClient *mesherykube.Client + // channelPool holds the fixed system channels (Stop/OS/ReSync) and is + // read-only after construction. Dynamic exec/log-stream sessions live in + // sessions (guarded by sessionsMu), not here, so the two never race. channelPool map[string]channels.GenericChannel + sessions map[string]channels.StructChannel + sessionsMu sync.Mutex stores map[string]cache.Store outputWriter output.Writer outputFiltration internalconfig.OutputFiltrationContainer @@ -65,6 +72,7 @@ func New( kubeClient: kubeClient, clusterID: clusterID, channelPool: pool, + sessions: make(map[string]channels.StructChannel), outputFiltration: outputFiltration, }, nil } diff --git a/meshsync/sessions.go b/meshsync/sessions.go new file mode 100644 index 00000000..71363c67 --- /dev/null +++ b/meshsync/sessions.go @@ -0,0 +1,57 @@ +package meshsync + +import ( + "github.com/meshery/meshsync/internal/channels" +) + +// Interactive exec and log-stream sessions are keyed by a per-request id. +// They were previously stored in the shared channelPool alongside the fixed +// system channels (Stop/OS/ReSync), which meant session goroutines mutated the +// same map that other goroutines read (system-channel selects, getActiveChannels), +// a data race that can panic the process. They now live in their own +// mutex-guarded map so channelPool stays read-only after initialization. +// +// Every helper returns the channel (if any) and releases the lock before the +// caller performs any channel send/receive, so the sessions mutex is never held +// across a blocking channel operation. + +// addSession registers a new session channel for id and returns it with +// created=true. If a session already exists for id, the existing channel is +// returned with created=false so the caller does not start a duplicate. +func (h *Handler) addSession(id string) (ch channels.StructChannel, created bool) { + h.sessionsMu.Lock() + defer h.sessionsMu.Unlock() + if existing, ok := h.sessions[id]; ok { + return existing, false + } + ch = channels.NewStructChannel() + h.sessions[id] = ch + return ch, true +} + +// getSession returns the session channel for id, if present. +func (h *Handler) getSession(id string) (channels.StructChannel, bool) { + h.sessionsMu.Lock() + defer h.sessionsMu.Unlock() + ch, ok := h.sessions[id] + return ch, ok +} + +// deleteSession removes the session for id. It is safe to call for an id that +// is not present. +func (h *Handler) deleteSession(id string) { + h.sessionsMu.Lock() + defer h.sessionsMu.Unlock() + delete(h.sessions, id) +} + +// activeSessionIDs returns the ids of the currently active sessions. +func (h *Handler) activeSessionIDs() []string { + h.sessionsMu.Lock() + defer h.sessionsMu.Unlock() + ids := make([]string, 0, len(h.sessions)) + for id := range h.sessions { + ids = append(ids, id) + } + return ids +} diff --git a/meshsync/sessions_test.go b/meshsync/sessions_test.go new file mode 100644 index 00000000..2d8e4db1 --- /dev/null +++ b/meshsync/sessions_test.go @@ -0,0 +1,71 @@ +package meshsync + +import ( + "fmt" + "sync" + "testing" + + "github.com/meshery/meshsync/internal/channels" +) + +func newSessionsHandler() *Handler { + return &Handler{sessions: make(map[string]channels.StructChannel)} +} + +func TestAddSessionIsIdempotent(t *testing.T) { + h := newSessionsHandler() + + ch1, created1 := h.addSession("a") + if !created1 { + t.Fatal("first addSession should report created=true") + } + ch2, created2 := h.addSession("a") + if created2 { + t.Fatal("second addSession for the same id should report created=false") + } + if ch1 != ch2 { + t.Fatal("addSession should return the same channel for an existing id") + } + if _, ok := h.getSession("a"); !ok { + t.Fatal("getSession should find the added session") + } + if got := h.activeSessionIDs(); len(got) != 1 || got[0] != "a" { + t.Fatalf("activeSessionIDs = %v, want [a]", got) + } + + h.deleteSession("a") + if _, ok := h.getSession("a"); ok { + t.Fatal("getSession should not find a deleted session") + } + if got := h.activeSessionIDs(); len(got) != 0 { + t.Fatalf("activeSessionIDs after delete = %v, want []", got) + } + // deleteSession on a missing id must be a no-op, not a panic. + h.deleteSession("missing") +} + +// TestSessionsConcurrentAccess must pass under -race: many goroutines add, read, +// enumerate, and delete overlapping session ids simultaneously. Before sessions +// were split out of channelPool, this shape of access was a concurrent map +// read/write. +func TestSessionsConcurrentAccess(t *testing.T) { + h := newSessionsHandler() + + const workers = 16 + const iterations = 500 + var wg sync.WaitGroup + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + id := fmt.Sprintf("session-%d", w%4) // deliberate overlap across workers + for i := 0; i < iterations; i++ { + h.addSession(id) + h.getSession(id) + h.activeSessionIDs() + h.deleteSession(id) + } + }(w) + } + wg.Wait() +} From a11f99d36f85a9d9b4fc1d45a26457607efee0a2 Mon Sep 17 00:00:00 2001 From: Lee Calcote Date: Sat, 4 Jul 2026 19:08:05 -0500 Subject: [PATCH 2/5] fix(meshsync): harden log-stream lifecycle (deadlock, goroutine leak, error loop) Three pre-existing issues in streamLogs, surfaced while moving sessions out of channelPool: - Stop requests now send to the session channel non-blockingly, so a duplicate stop or an already-closing stream cannot freeze the log-request loop. - The waiter goroutine now selects on a done channel closed when streamLogs exits, so it no longer leaks when the stream ends on its own (EOF/error). Session removal and stream close move to defers covering every exit path. - A non-EOF read error now breaks the loop (checked before the zero-byte continue), avoiding an infinite read/log loop on a failed stream. Addresses review feedback on #587. Signed-off-by: marblom007 <158522975+marblom007@users.noreply.github.com> Signed-off-by: Lee Calcote --- meshsync/logstream.go | 51 +++++++++++++++++++++++++++++-------------- 1 file changed, 35 insertions(+), 16 deletions(-) diff --git a/meshsync/logstream.go b/meshsync/logstream.go index 2d2627e6..a7684295 100644 --- a/meshsync/logstream.go +++ b/meshsync/logstream.go @@ -28,8 +28,13 @@ func (h *Handler) processLogRequest(obj interface{}, cfg config.ListenerConfig) id := fmt.Sprintf("logs.%s.%s.%s", req.Namespace, req.Name, req.Container) if bool(req.Stop) { // Stop request: signal the running stream, if any, to close. + // Non-blocking: the stream may already be closing and no longer + // receiving, so a plain send could freeze this loop. if ch, ok := h.getSession(id); ok { - ch <- struct{}{} + select { + case ch <- struct{}{}: + default: + } } continue } @@ -42,6 +47,10 @@ func (h *Handler) processLogRequest(obj interface{}, cfg config.ListenerConfig) } func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.ListenerConfig) { + // Remove the session however this function exits (stream-open failure, EOF, + // read error, or a stop request). + defer h.deleteSession(id) + resp, err := h.kubeClient.KubeClient.CoreV1().Pods(req.Namespace).GetLogs(req.Name, &v1.PodLogOptions{ Container: req.Container, Follow: req.Follow, @@ -55,17 +64,27 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene }).Stream(context.TODO()) if err != nil { h.Log.Error(ErrLogStream(err)) - h.deleteSession(id) return } + defer resp.Close() + + // done unblocks the waiter goroutine when the stream ends on its own (EOF or + // read error), so it is not leaked once streamLogs returns. + done := make(chan struct{}) + defer close(done) go func() { - if ch, ok := h.getSession(id); ok { - <-ch + ch, ok := h.getSession(id) + if !ok { + return + } + select { + case <-ch: + // Stop request: close the stream so the read loop unblocks and exits. + h.Log.Debugf("Closing %s", id) + resp.Close() + case <-done: } - h.Log.Debugf("Closing %s", id) - h.deleteSession(id) - resp.Close() }() for { @@ -74,16 +93,18 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene if err == io.EOF { break } - if numBytes == 0 { - continue - } if err != nil { + // A non-EOF read error ends the stream; breaking avoids an infinite + // read/log loop on a failed stream. h.Log.Error(ErrCopyBuffer(err)) - h.deleteSession(id) + break + } + if numBytes == 0 { + continue } message := string(buf[:numBytes]) - err = h.Broker.Publish(cfg.PublishTo, &broker.Message{ + if pubErr := h.Broker.Publish(cfg.PublishTo, &broker.Message{ ObjectType: broker.LogStreamObject, EventType: broker.Add, Object: &model.LogObject{ @@ -92,10 +113,8 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene Primary: req.Name, Secondary: req.Container, }, - }) - if err != nil { - h.Log.Error(ErrCopyBuffer(err)) + }); pubErr != nil { + h.Log.Error(ErrCopyBuffer(pubErr)) } } - } From 2a53afb2d47545a9e2fab7ff5efb970686a6b779 Mon Sep 17 00:00:00 2001 From: Lee Calcote Date: Sun, 5 Jul 2026 19:29:52 -0500 Subject: [PATCH 3/5] fix(meshsync): buffer session channels and guard nil sessions map Session stop signals are sent non-blocking from exec/log-stream, so an unbuffered session channel dropped a stop that arrived before the session goroutine started receiving, leaving the session running. Session channels are now 1-buffered so a single stop is reliably recorded. addSession also lazily initializes the sessions map so a Handler built outside New does not panic on the write. Adds regression tests. Addresses review feedback on #587. Signed-off-by: Lee Calcote --- meshsync/sessions.go | 10 +++++++++- meshsync/sessions_test.go | 29 +++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/meshsync/sessions.go b/meshsync/sessions.go index 71363c67..0db5f8ed 100644 --- a/meshsync/sessions.go +++ b/meshsync/sessions.go @@ -21,10 +21,18 @@ import ( func (h *Handler) addSession(id string) (ch channels.StructChannel, created bool) { h.sessionsMu.Lock() defer h.sessionsMu.Unlock() + // Defensive: a Handler constructed outside New has a nil map; writing to it + // would panic. + if h.sessions == nil { + h.sessions = make(map[string]channels.StructChannel) + } if existing, ok := h.sessions[id]; ok { return existing, false } - ch = channels.NewStructChannel() + // 1-buffered: exec/log-stream stop signals are sent non-blocking, so a stop + // that arrives before the session goroutine begins receiving must still be + // recorded rather than dropped (which would leave the session running). + ch = make(channels.StructChannel, 1) h.sessions[id] = ch return ch, true } diff --git a/meshsync/sessions_test.go b/meshsync/sessions_test.go index 2d8e4db1..72423129 100644 --- a/meshsync/sessions_test.go +++ b/meshsync/sessions_test.go @@ -69,3 +69,32 @@ func TestSessionsConcurrentAccess(t *testing.T) { } wg.Wait() } + +func TestAddSessionChannelIsBuffered(t *testing.T) { + h := newSessionsHandler() + ch, created := h.addSession("s") + if !created { + t.Fatal("expected the session to be created") + } + // The session channel must be buffered: exec/log-stream send stop signals + // non-blocking, so a stop that arrives before the receiver is ready must land + // in the buffer rather than being dropped. + select { + case ch <- struct{}{}: + default: + t.Fatal("session channel is unbuffered: a non-blocking stop signal was dropped") + } +} + +func TestAddSessionInitializesNilMap(t *testing.T) { + // A Handler constructed outside New has a nil sessions map; addSession must + // initialize it rather than panic on the write. + h := &Handler{} + ch, created := h.addSession("s") + if !created || ch == nil { + t.Fatal("addSession should lazily initialize the map and create the session") + } + if _, ok := h.getSession("s"); !ok { + t.Fatal("session should be retrievable after lazy initialization") + } +} From ec117fab333d195e5333b08d4091607c95c5fd1a Mon Sep 17 00:00:00 2001 From: Lee Calcote Date: Sun, 5 Jul 2026 23:07:02 -0500 Subject: [PATCH 4/5] fix(meshsync): terminate log-stream sessions on global stop The log-stream waiter goroutine now also selects on the global channels.Stop channel, so a long-running log stream is closed promptly on shutdown instead of blocking graceful termination until the stream ends on its own (matching the exec session loop). channelPool holds only the fixed system channels and is read-only after init, so the read needs no lock. Addresses review feedback on #587. Signed-off-by: Lee Calcote --- meshsync/logstream.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/meshsync/logstream.go b/meshsync/logstream.go index a7684295..5290c5f2 100644 --- a/meshsync/logstream.go +++ b/meshsync/logstream.go @@ -7,6 +7,7 @@ import ( "io" "github.com/meshery/meshkit/broker" + "github.com/meshery/meshsync/internal/channels" "github.com/meshery/meshsync/internal/config" "github.com/meshery/meshsync/pkg/model" v1 "k8s.io/api/core/v1" @@ -83,6 +84,12 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene // Stop request: close the stream so the read loop unblocks and exits. h.Log.Debugf("Closing %s", id) resp.Close() + case <-h.channelPool[channels.Stop].(channels.StopChannel): + // Global shutdown: close the stream so a long-running log stream does + // not block graceful shutdown. channelPool holds only the fixed system + // channels and is read-only after init, so this read needs no lock. + h.Log.Debugf("Stopping session %s on global stop", id) + resp.Close() case <-done: } }() From 4c6c25d25f2ffbcd35cb7a70dc394dc219b2ff79 Mon Sep 17 00:00:00 2001 From: Lee Calcote Date: Sun, 5 Jul 2026 23:10:45 -0500 Subject: [PATCH 5/5] refactor(meshsync): extract log-stream stop waiter to reduce complexity Move the waiter goroutine's select into awaitLogStreamStop so streamLogs stays under the cyclomatic-complexity limit (golangci-lint cyclop) after adding the global-stop case. No behavior change. Signed-off-by: Lee Calcote --- meshsync/logstream.go | 41 ++++++++++++++++++++++------------------- 1 file changed, 22 insertions(+), 19 deletions(-) diff --git a/meshsync/logstream.go b/meshsync/logstream.go index 5290c5f2..8556f56d 100644 --- a/meshsync/logstream.go +++ b/meshsync/logstream.go @@ -74,25 +74,7 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene done := make(chan struct{}) defer close(done) - go func() { - ch, ok := h.getSession(id) - if !ok { - return - } - select { - case <-ch: - // Stop request: close the stream so the read loop unblocks and exits. - h.Log.Debugf("Closing %s", id) - resp.Close() - case <-h.channelPool[channels.Stop].(channels.StopChannel): - // Global shutdown: close the stream so a long-running log stream does - // not block graceful shutdown. channelPool holds only the fixed system - // channels and is read-only after init, so this read needs no lock. - h.Log.Debugf("Stopping session %s on global stop", id) - resp.Close() - case <-done: - } - }() + go h.awaitLogStreamStop(id, resp, done) for { buf := make([]byte, 2000) @@ -125,3 +107,24 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene } } } + +// awaitLogStreamStop closes resp when the session receives an explicit stop +// signal or the process is shutting down (channels.Stop), and returns without +// closing when the stream has already ended on its own (done is closed by +// streamLogs). channelPool holds only the fixed system channels and is +// read-only after init, so reading channels.Stop from it needs no lock. +func (h *Handler) awaitLogStreamStop(id string, resp io.ReadCloser, done <-chan struct{}) { + ch, ok := h.getSession(id) + if !ok { + return + } + select { + case <-ch: + h.Log.Debugf("Closing %s", id) + resp.Close() + case <-h.channelPool[channels.Stop].(channels.StopChannel): + h.Log.Debugf("Stopping session %s on global stop", id) + resp.Close() + case <-done: + } +}