From 510bee2f30130e0859667316d844eeebfab08315 Mon Sep 17 00:00:00 2001
From: Sanaei <ho3ein.sanaei@gmail.com>
Date: Fri, 6 Nov 2026 18:47:13 +0100
Subject: [PATCH] fix(traffic): serialize inline traffic writes with the writer

submitTrafficWrite falls back to running fn inline when the serial
writer is not running (before StartTrafficWriter, during the shutdown
drain, in CLI paths and tests). That path took no lock, so an inline
write could run beside the draining writer or another inline write,
breaking the single-writer invariant that guards the Postgres deadlock.

One visible effect: ResetTrafficByEmail fans out one reset per inbound.
Run side by side, the second reset read the record still disabled but
the traffic already zeroed, judged the client operator-disabled and
saved enable=false over the first reset's restore
(TestResetTrafficByEmail_ReenablesEveryInboundOfTheClient failed ~25%).

safeApply now holds a package mutex, so every write path is serialized.
---
 internal/web/service/traffic_writer.go      |  5 +++
 internal/web/service/traffic_writer_test.go | 49 +++++++++++++++++++++
 2 files changed, 54 insertions(+)

diff --git a/internal/web/service/traffic_writer.go b/internal/web/service/traffic_writer.go
index b65c9eea5..a40b17484 100644
--- a/internal/web/service/traffic_writer.go
+++ b/internal/web/service/traffic_writer.go
@@ -32,6 +32,7 @@ var (
 	twCtx        context.Context
 	twCancel     context.CancelFunc
 	twDone       chan struct{}
+	twApplyMu    sync.Mutex
 )
 
 // StartTrafficWriter spins up the serial writer goroutine. Safe to call again
@@ -197,7 +198,11 @@ func isSerializedTx(tx *gorm.DB) bool {
 	return serializedTxOf(tx) != nil
 }
 
+// safeApply holds twApplyMu so an inline fallback (no writer, or shutdown drain)
+// never runs beside another write: overlapping resets switched a client back off.
 func safeApply(fn func() error) (err error) {
+	twApplyMu.Lock()
+	defer twApplyMu.Unlock()
 	defer func() {
 		if r := recover(); r != nil {
 			err = fmt.Errorf("traffic writer panic: %v", r)
diff --git a/internal/web/service/traffic_writer_test.go b/internal/web/service/traffic_writer_test.go
index 8623cf4b1..35189cd28 100644
--- a/internal/web/service/traffic_writer_test.go
+++ b/internal/web/service/traffic_writer_test.go
@@ -63,6 +63,55 @@ func TestTrafficWriterSubmitAfterStopRunsInline(t *testing.T) {
 	}
 }
 
+// The inline fallback ran beside other writes, so two resets of one client
+// raced and the second one switched the client back off.
+func TestTrafficWriterInlineFallbackNeverOverlapsAnotherWrite(t *testing.T) {
+	for _, tt := range []struct {
+		name  string
+		setup func(t *testing.T, first func())
+	}{
+		{"writer never started", func(t *testing.T, first func()) {
+			go func() { _ = submitTrafficWrite(func() error { first(); return nil }) }()
+		}},
+		{"submitted while shutdown drains", func(t *testing.T, first func()) {
+			StartTrafficWriter()
+			go func() { _ = submitTrafficWrite(func() error { first(); return nil }) }()
+			twMu.Lock()
+			ctx := twCtx
+			twMu.Unlock()
+			go StopTrafficWriter()
+			<-ctx.Done()
+		}},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			resetTrafficWriterForTest(t)
+			entered, secondStarted := make(chan struct{}), make(chan struct{})
+			var inFirst, overlapped atomic.Bool
+			tt.setup(t, func() {
+				inFirst.Store(true)
+				close(entered)
+				select {
+				case <-secondStarted:
+				case <-time.After(300 * time.Millisecond):
+				}
+				inFirst.Store(false)
+			})
+			<-entered
+			err := submitTrafficWrite(func() error {
+				overlapped.Store(inFirst.Load())
+				close(secondStarted)
+				return nil
+			})
+			if err != nil {
+				t.Fatalf("second submitTrafficWrite: %v", err)
+			}
+			if overlapped.Load() {
+				t.Fatal("inline write ran while another write was still applying")
+			}
+		})
+	}
+}
+
 func TestTrafficWriterStopDrainsQueuedWrite(t *testing.T) {
 	resetTrafficWriterForTest(t)