fix(history): detach worker query context from shutdown cancellation (#420)

The history worker reused one long-lived cancellable context as the parent for every periodic database/sql flush while Close cancelled that same parent on another goroutine. That derive-vs-cancel contact on the shared parent intermittently crashed in context.propagateCancel (concurrent map writes, #408).

commit: 461ba1082c3bcbdb5a840a39e5ff4a901cac3681
author: bnema · 43d ago · 2026-09-09 05:51 UTC
committer: GitHub · 43d ago · 2026-09-09 05:51 UTC
parent: 5fc5c8e
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

2 files · +66 -14

M internal/application/usecase/history_recorder.go +13 -2


@@ -355,6 +355,14 @@ func (uc *HistoryRecorderUseCase) historyWorker() {
355	355		ticker := time.NewTicker(historyWorkerFlushInterval)
356	356		defer ticker.Stop()
357	357		workerCtx := uc.workerContext()
	358	+	// queryCtx carries the worker's values (logging) but is detached from
	359	+	// worker cancellation. Every database/sql query derives per-query child
	360	+	// contexts from its parent; deriving those children from workerCtx while
	361	+	// Close cancels workerCtx intermittently crashed in
	362	+	// context.propagateCancel (concurrent map writes, #408). Shutdown
	363	+	// coordination stays on uc.done/workerCtx.Done() in the select below,
	364	+	// and the shutdown drain uses its own bounded timeout context.
	365	+	queryCtx := context.WithoutCancel(workerCtx)
358	366
359	367		pending := newPendingHistoryRecords()
360	368		for {

@@ -362,7 +370,7 @@ func (uc *HistoryRecorderUseCase) historyWorker() {
362	370		case record := <-uc.historyQueue:
363	371		pending.add(record)
364	372		case <-ticker.C:
365		-	_ = uc.flushPendingHistory(workerCtx, pending)
	373	+	_ = uc.flushPendingHistory(queryCtx, pending)
366	374		case req := <-uc.controlQueue:
367	375		uc.handleHistoryControl(pending, req)
368	376		case <-uc.done:

@@ -598,7 +606,10 @@ func (uc *HistoryRecorderUseCase) publishHistoryChange(successfulVisits, success
598	606		reasons = append(reasons, dto.HistoryChangeReasonTitle)
599	607		}
600	608		sink := uc.historyChangeSink()
601		-	sink.OnHistoryChanged(uc.workerContext(), dto.HistoryChange{
	609	+	// Detach the notification from worker cancellation: the sink runs on
	610	+	// another goroutine (main-thread dispatch) and must never derive from
	611	+	// or observe a parent that Close cancels concurrently (see #408).
	612	+	sink.OnHistoryChanged(context.WithoutCancel(uc.workerContext()), dto.HistoryChange{
602	613		Reasons: reasons,
603	614		VisitCount: successfulVisits,
604	615		TitleCount: successfulTitles,

 	ticker := time.NewTicker(historyWorkerFlushInterval)
 	defer ticker.Stop()
 	workerCtx := uc.workerContext()
+	// queryCtx carries the worker's values (logging) but is detached from
+	// worker cancellation. Every database/sql query derives per-query child
+	// contexts from its parent; deriving those children from workerCtx while
+	// Close cancels workerCtx intermittently crashed in
+	// context.propagateCancel (concurrent map writes, #408). Shutdown
+	// coordination stays on uc.done/workerCtx.Done() in the select below,
+	// and the shutdown drain uses its own bounded timeout context.
+	queryCtx := context.WithoutCancel(workerCtx)
 	pending := newPendingHistoryRecords()
 	for {
 		case record := <-uc.historyQueue:
 			pending.add(record)
 		case <-ticker.C:
-			_ = uc.flushPendingHistory(workerCtx, pending)
+			_ = uc.flushPendingHistory(queryCtx, pending)
 		case req := <-uc.controlQueue:
 			uc.handleHistoryControl(pending, req)
 		case <-uc.done:
 		reasons = append(reasons, dto.HistoryChangeReasonTitle)
 	}
 	sink := uc.historyChangeSink()
-	sink.OnHistoryChanged(uc.workerContext(), dto.HistoryChange{
+	// Detach the notification from worker cancellation: the sink runs on
+	// another goroutine (main-thread dispatch) and must never derive from
+	// or observe a parent that Close cancels concurrently (see #408).
+	sink.OnHistoryChanged(context.WithoutCancel(uc.workerContext()), dto.HistoryChange{
 		Reasons:    reasons,
 		VisitCount: successfulVisits,
 		TitleCount: successfulTitles,

M internal/application/usecase/history_recorder_test.go +53 -12


@@ -33,6 +33,29 @@ func (s *recordingHistoryChangeSink) snapshot() []dto.HistoryChange {
33	33		return out
34	34		}
35	35
	36	+	// ctxCapturingHistoryChangeSink records the contexts delivered with each
	37	+	// change so tests can assert detachment from worker cancellation (#408).
	38	+	type ctxCapturingHistoryChangeSink struct {
	39	+	recordingHistoryChangeSink
	40	+	mu sync.Mutex
	41	+	ctxs []context.Context
	42	+	}
	43	+
	44	+	func (s *ctxCapturingHistoryChangeSink) OnHistoryChanged(ctx context.Context, change dto.HistoryChange) {
	45	+	s.mu.Lock()
	46	+	s.ctxs = append(s.ctxs, ctx)
	47	+	s.mu.Unlock()
	48	+	s.recordingHistoryChangeSink.OnHistoryChanged(ctx, change)
	49	+	}
	50	+
	51	+	func (s *ctxCapturingHistoryChangeSink) snapshotCtxs() []context.Context {
	52	+	s.mu.Lock()
	53	+	defer s.mu.Unlock()
	54	+	out := make([]context.Context, len(s.ctxs))
	55	+	copy(out, s.ctxs)
	56	+	return out
	57	+	}
	58	+
36	59		func TestNormalizeHistoryChangeSink_TreatsTypedNilAsNoop(t *testing.T) {
37	60		var sink *recordingHistoryChangeSink
38	61

@@ -204,26 +227,29 @@ func TestHistoryRecorder_CloseDrainsPendingAndPublishes(t *testing.T) {
204	227		require.Equal(t, 1, changes[0].VisitCount)
205	228		}
206	229
207		-	func TestHistoryRecorder_CloseCancelsInFlightWorkerFlushAndDrains(t *testing.T) {
	230	+	func TestHistoryRecorder_PeriodicFlushUsesDetachedQueryContext(t *testing.T) {
208	231		ctx := context.Background()
209		-	sink := &recordingHistoryChangeSink{}
	232	+	sink := &ctxCapturingHistoryChangeSink{}
210	233		repo := repomocks.NewMockHistoryRepository(t)
211		-	const historyURL = "https://example.com/cancel-close"
	234	+	const historyURL = "https://example.com/detached-flush"
212	235
213	236		findStarted := make(chan struct{})
	237	+	releaseFind := make(chan struct{})
214	238		var findOnce sync.Once
	239	+	var captureMu sync.Mutex
	240	+	var captured context.Context
215	241		repo.EXPECT().FindByURL(mock.Anything, historyURL).RunAndReturn(
216	242		func(ctx context.Context, _ string) (*entity.HistoryEntry, error) {
217		-	first := false
218		-	findOnce.Do(func() { first = true })
219		-	if first {
220		-	close(findStarted)
221		-	<-ctx.Done()
222		-	return nil, ctx.Err()
	243	+	findOnce.Do(func() { close(findStarted) })
	244	+	captureMu.Lock()
	245	+	if captured == nil {
	246	+	captured = ctx
223	247		}
	248	+	captureMu.Unlock()
	249	+	<-releaseFind
224	250		return nil, nil
225	251		},
226		-	).Twice()
	252	+	).Once()
227	253		repo.EXPECT().Save(mock.Anything, mock.MatchedBy(func(entry *entity.HistoryEntry) bool {
228	254		return entry.URL == historyURL && entry.VisitCount == 1
229	255		})).Return(nil).Once()

@@ -233,22 +259,37 @@ func TestHistoryRecorder_CloseCancelsInFlightWorkerFlushAndDrains(t *testing.T)
233	259
234	260		select {
235	261		case <-findStarted:
236		-	case <-time.After(time.Second):
	262	+	case <-time.After(2 * time.Second):
237	263		t.Fatal("timed out waiting for periodic history flush to start")
238	264		}
239	265
	266	+	// Close must not cancel the in-flight periodic flush: the flush runs to
	267	+	// completion with a context detached from worker cancellation, and the
	268	+	// shutdown drain then finds nothing left to persist.
240	269		closed := make(chan struct{})
241	270		go func() {
242	271		uc.Close()
243	272		close(closed)
244	273		}()
	274	+	close(releaseFind)
245	275
246	276		select {
247	277		case <-closed:
248		-	case <-time.After(time.Second):
	278	+	case <-time.After(5 * time.Second):
249	279		t.Fatal("timed out waiting for history recorder shutdown")
250	280		}
251	281
	282	+	captureMu.Lock()
	283	+	defer captureMu.Unlock()
	284	+	require.NotNil(t, captured, "periodic flush must reach the repository")
	285	+	require.Nil(t, captured.Done(),
	286	+	"periodic flush query context must be detached from worker cancellation (#408)")
	287	+	for _, sinkCtx := range sink.snapshotCtxs() {
	288	+	require.NotNil(t, sinkCtx)
	289	+	require.Nil(t, sinkCtx.Done(),
	290	+	"history change notification must be detached from worker cancellation (#408)")
	291	+	}
	292	+
252	293		changes := sink.snapshot()
253	294		require.Len(t, changes, 1)
254	295		require.Equal(t, 1, changes[0].VisitCount)

 	return out
 }
+// ctxCapturingHistoryChangeSink records the contexts delivered with each
+// change so tests can assert detachment from worker cancellation (#408).
+type ctxCapturingHistoryChangeSink struct {
+	recordingHistoryChangeSink
+	mu   sync.Mutex
+	ctxs []context.Context
+}
+func (s *ctxCapturingHistoryChangeSink) OnHistoryChanged(ctx context.Context, change dto.HistoryChange) {
+	s.mu.Lock()
+	s.ctxs = append(s.ctxs, ctx)
+	s.mu.Unlock()
+	s.recordingHistoryChangeSink.OnHistoryChanged(ctx, change)
+}
+func (s *ctxCapturingHistoryChangeSink) snapshotCtxs() []context.Context {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]context.Context, len(s.ctxs))
+	copy(out, s.ctxs)
+	return out
+}
 func TestNormalizeHistoryChangeSink_TreatsTypedNilAsNoop(t *testing.T) {
 	var sink *recordingHistoryChangeSink
 	require.Equal(t, 1, changes[0].VisitCount)
 }
-func TestHistoryRecorder_CloseCancelsInFlightWorkerFlushAndDrains(t *testing.T) {
+func TestHistoryRecorder_PeriodicFlushUsesDetachedQueryContext(t *testing.T) {
 	ctx := context.Background()
-	sink := &recordingHistoryChangeSink{}
+	sink := &ctxCapturingHistoryChangeSink{}
 	repo := repomocks.NewMockHistoryRepository(t)
-	const historyURL = "https://example.com/cancel-close"
+	const historyURL = "https://example.com/detached-flush"
 	findStarted := make(chan struct{})
+	releaseFind := make(chan struct{})
 	var findOnce sync.Once
+	var captureMu sync.Mutex
+	var captured context.Context
 	repo.EXPECT().FindByURL(mock.Anything, historyURL).RunAndReturn(
 		func(ctx context.Context, _ string) (*entity.HistoryEntry, error) {
-			first := false
+			findOnce.Do(func() { close(findStarted) })
-			findOnce.Do(func() { first = true })
+			captureMu.Lock()
-			if first {
+			if captured == nil {
-				close(findStarted)
+				captured = ctx
-				<-ctx.Done()
-				return nil, ctx.Err()
 			}
+			captureMu.Unlock()
+			<-releaseFind
 			return nil, nil
 		},
-	).Twice()
+	).Once()
 	repo.EXPECT().Save(mock.Anything, mock.MatchedBy(func(entry *entity.HistoryEntry) bool {
 		return entry.URL == historyURL && entry.VisitCount == 1
 	})).Return(nil).Once()
 	select {
 	case <-findStarted:
-	case <-time.After(time.Second):
+	case <-time.After(2 * time.Second):
 		t.Fatal("timed out waiting for periodic history flush to start")
 	}
+	// Close must not cancel the in-flight periodic flush: the flush runs to
+	// completion with a context detached from worker cancellation, and the
+	// shutdown drain then finds nothing left to persist.
 	closed := make(chan struct{})
 	go func() {
 		uc.Close()
 		close(closed)
 	}()
+	close(releaseFind)
 	select {
 	case <-closed:
-	case <-time.After(time.Second):
+	case <-time.After(5 * time.Second):
 		t.Fatal("timed out waiting for history recorder shutdown")
 	}
+	captureMu.Lock()
+	defer captureMu.Unlock()
+	require.NotNil(t, captured, "periodic flush must reach the repository")
+	require.Nil(t, captured.Done(),
+		"periodic flush query context must be detached from worker cancellation (#408)")
+	for _, sinkCtx := range sink.snapshotCtxs() {
+		require.NotNil(t, sinkCtx)
+		require.Nil(t, sinkCtx.Done(),
+			"history change notification must be detached from worker cancellation (#408)")
+	}
 	changes := sink.snapshot()
 	require.Len(t, changes, 1)
 	require.Equal(t, 1, changes[0].VisitCount)