From bc9025d5660399a84d15261f18d92d446e74e610 Mon Sep 17 00:00:00 2001
From: lingniu <lingniu@lingnius-Mac-mini.local>
Date: Thu, 2 Jul 2026 13:25:15 +0800
Subject: [PATCH] fix(go): protect durable spool replay during shutdown

---
 .../internal/eventbus/durable_sink.go         | 10 +++-
 .../internal/eventbus/durable_sink_test.go    | 46 +++++++++++++++++++
 2 files changed, 55 insertions(+), 1 deletion(-)

diff --git a/go/vehicle-gateway/internal/eventbus/durable_sink.go b/go/vehicle-gateway/internal/eventbus/durable_sink.go
index a7f0940a..eab71a14 100644
--- a/go/vehicle-gateway/internal/eventbus/durable_sink.go
+++ b/go/vehicle-gateway/internal/eventbus/durable_sink.go
@@ -37,6 +37,8 @@ type durableRecordFile struct {
 	record durableRecord
 }
 
+const durableReplayOperationTimeout = 30 * time.Second
+
 func NewDurableSink(delegate Sink, cfg DurableConfig) *DurableSink {
 	if delegate == nil {
 		panic("durable delegate sink must not be nil")
@@ -130,13 +132,19 @@ func (s *DurableSink) ReplayLoop(ctx context.Context, interval time.Duration, on
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			if err := s.ReplayOnce(ctx); err != nil && onError != nil {
+			if err := s.replayOnceFromLoop(ctx); err != nil && onError != nil {
 				onError(err)
 			}
 		}
 	}
 }
 
+func (s *DurableSink) replayOnceFromLoop(ctx context.Context) error {
+	replayCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), durableReplayOperationTimeout)
+	defer cancel()
+	return s.ReplayOnce(replayCtx)
+}
+
 func (s *DurableSink) Close() error {
 	return s.delegate.Close()
 }
diff --git a/go/vehicle-gateway/internal/eventbus/durable_sink_test.go b/go/vehicle-gateway/internal/eventbus/durable_sink_test.go
index b2c0a1b8..cdf508c3 100644
--- a/go/vehicle-gateway/internal/eventbus/durable_sink_test.go
+++ b/go/vehicle-gateway/internal/eventbus/durable_sink_test.go
@@ -95,6 +95,29 @@ func TestDurableSinkReplayPublishesRawBeforeUnifiedEvenWhenFilesAreOutOfOrder(t
 	}
 }
 
+func TestDurableSinkReplayLoopTickUsesUncancelledContext(t *testing.T) {
+	dir := t.TempDir()
+	env := durableTestEnvelope()
+	writeDurableRecord(t, filepath.Join(dir, "0001-raw.json"), durableRecord{Kind: "raw", Envelope: env})
+	delegate := &contextCheckingReplaySink{}
+	sink := NewDurableSink(delegate, DurableConfig{Directory: dir})
+	parent, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	if err := sink.replayOnceFromLoop(parent); err != nil {
+		t.Fatalf("replayOnceFromLoop() error = %v", err)
+	}
+	if delegate.rawCtxErr != nil {
+		t.Fatalf("delegate saw cancelled context: %v", delegate.rawCtxErr)
+	}
+	if delegate.rawCalls != 1 {
+		t.Fatalf("raw calls = %d, want 1", delegate.rawCalls)
+	}
+	if files := spoolFiles(t, dir); len(files) != 0 {
+		t.Fatalf("spool files after replay = %#v, want none", files)
+	}
+}
+
 func durableTestEnvelope() envelope.FrameEnvelope {
 	return envelope.FrameEnvelope{
 		Protocol:     envelope.ProtocolJT808,
@@ -187,3 +210,26 @@ func (s *scriptedSink) PublishUnified(context.Context, envelope.FrameEnvelope) e
 func (s *scriptedSink) Close() error {
 	return nil
 }
+
+type contextCheckingReplaySink struct {
+	rawCtxErr     error
+	unifiedCtxErr error
+	rawCalls      int
+	unifiedCalls  int
+}
+
+func (s *contextCheckingReplaySink) PublishRaw(ctx context.Context, _ envelope.FrameEnvelope) error {
+	s.rawCtxErr = ctx.Err()
+	s.rawCalls++
+	return s.rawCtxErr
+}
+
+func (s *contextCheckingReplaySink) PublishUnified(ctx context.Context, _ envelope.FrameEnvelope) error {
+	s.unifiedCtxErr = ctx.Err()
+	s.unifiedCalls++
+	return s.unifiedCtxErr
+}
+
+func (s *contextCheckingReplaySink) Close() error {
+	return nil
+}