fix(egress): retry mitmdump restart with backoff instead of giving up (#942)

* fix(egress): retry mitmdump restart with backoff instead of giving up

Previously, if Launch or WaitListenPort failed during a restart (e.g.
under node memory pressure that just OOM-killed mitmdump), the watchdog
goroutine would log "giving up" and return, leaving egress in a silent
dead state with no future restarts.

Replace the one-shot restart with restartWithBackoff: retry forever with
exponential backoff (1s -> 30s), kill half-launched processes, drain
stale exit signals on success, and respect ctx cancellation. Readiness
gate stays false across the retry window so k8s drains traffic.

* fix(egress): tag mitmdump exits with generation to avoid losing real death

The previous restart-with-backoff fix drained restartCh after a successful
relaunch to discard stale events from killed half-launched attempts. That
drain has a race: if the freshly-restarted mitmdump dies between
WaitListenPort returning and the drain executing, the real death event is
swallowed and the watcher returns to the outer loop with nothing pending,
re-entering the silent dead state the original fix was meant to prevent.

Tag every Launch with a monotonic generation captured in its OnExit
closure. The watcher compares the event's generation against the currently
live generation (set atomically with setRunning) and ignores stale events
without draining. Real deaths of the current mitmdump always match the
live generation and trigger a restart.

* fix(egress): widen restartCh buffer and use GracefulShutdown on retry

Two follow-ups to the generation-tagged watchdog:

1. restartCh buffer was 1, so under a retry storm a single stale exit
   event from a half-launched-and-killed mitmdump occupies the slot.
   When a later attempt succeeds and the freshly-restarted mitmdump
   dies immediately (continued OOM pressure), its OnExit hits the
   default branch in launchTagged and the real death event is dropped.
   Watcher then reads the stale event, ignores it on gen mismatch, and
   blocks forever — the same silent-dead-state the watchdog is meant
   to prevent. Buffer is bumped to 64; stale events are still cheap to
   discard via the gen check, we just need room to hold them.

2. Replace direct Process.Kill on the half-launched mitmdump with
   mitmproxy.GracefulShutdown(_, 1s). Kill returns immediately, so the
   next attempt's Launch can race the dying process for the listen
   port and fail WaitListenPort purely on contention; GracefulShutdown
   waits for reap and is consistent with shutdown.go.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(egress): block OnExit send with shutdown escape so live death events are not dropped

The previous default-drop send in launchTagged could still lose the only
exit signal for the currently-live mitmdump under sustained retry storms:
once the buffer fills with stale events from killed half-launched
attempts, a fresh process's death event hits the default branch and the
watcher sees only stale generations, leaving the watchdog idle in the
exact silent-dead-state it is meant to prevent.

Switch OnExit to a blocking send guarded by a shutdownCh that is closed
when watchMitmproxy's ctx is cancelled. The watcher always drains the
channel, so blocking is bounded; on shutdown the escape branch fires and
we log a warning so any drop is observable. Buffer stays at 64 purely as
a perf cushion against goroutine pile-up during retry storms; correctness
no longer depends on the size.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

Author: Pangjiping

components/egress/mitmproxy_transparent.go

@@ -20,6 +20,7 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/alibaba/opensandbox/egress/pkg/constants"
@@ -29,13 +30,25 @@ import (
 	"github.com/alibaba/opensandbox/internal/safego"
 )
 
+// exitEvent carries an OnExit notification tagged with the generation of the
+// mitmdump process that produced it. Generation tagging lets the watcher tell
+// "the currently-running mitmdump just died" apart from "a half-launched
+// attempt we already killed during a retry storm just finished reaping".
+type exitEvent struct {
+	gen uint64
+	err error
+}
+
 type mitmTransparent struct {
-	mu        sync.Mutex
-	running   *mitmproxy.Running
-	port      int
-	uid       uint32
-	cfg       mitmproxy.Config
-	restartCh chan error
+	mu         sync.Mutex
+	running    *mitmproxy.Running
+	currentGen uint64 // generation of the mitmdump currently considered live
+	port       int
+	uid        uint32
+	cfg        mitmproxy.Config // OnExit must NOT be set here; built per-Launch
+	nextGen    uint64           // atomic; monotonic gen counter handed to each Launch
+	restartCh  chan exitEvent
+	shutdownCh chan struct{} // closed by watchMitmproxy on ctx cancel; lets OnExit unblock during shutdown
 }
 
 func (m *mitmTransparent) getRunning() *mitmproxy.Running {
@@ -44,10 +57,38 @@ func (m *mitmTransparent) getRunning() *mitmproxy.Running {
 	return m.running
 }
 
-func (m *mitmTransparent) setRunning(r *mitmproxy.Running) {
+func (m *mitmTransparent) setRunning(r *mitmproxy.Running, gen uint64) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.running = r
+	m.currentGen = gen
+}
+
+func (m *mitmTransparent) getCurrentGen() uint64 {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.currentGen
+}
+
+// launchTagged starts mitmdump with an OnExit closure that publishes the death
+// of this specific process (identified by gen) into restartCh.
+//
+// The send is blocking with shutdownCh as the only escape: dropping an exit
+// event while the watcher is still running can leave egress in a silent dead
+// state (the watcher would never see the death and never trigger a restart).
+// Stale events from killed half-launched attempts are still cheap to discard
+// downstream via the gen check in watchMitmproxy; we just must not lose them
+// in transit. Shutdown is the only legitimate reason to give up on a send,
+// and we log a warning when that happens so the drop is observable.
+func launchTagged(cfg mitmproxy.Config, restartCh chan<- exitEvent, shutdownCh <-chan struct{}, gen uint64) (*mitmproxy.Running, error) {
+	cfg.OnExit = func(err error) {
+		select {
+		case restartCh <- exitEvent{gen: gen, err: err}:
+		case <-shutdownCh:
+			log.Warnf("[mitmproxy] dropping exit event during shutdown (gen=%d): %v", gen, err)
+		}
+	}
+	return mitmproxy.Launch(cfg)
 }
 
 // startMitmproxyTransparentIfEnabled starts mitmdump in transparent mode, waits for the listener, and installs OUTPUT REDIRECT, then syncs the CA.
@@ -68,14 +109,15 @@ func startMitmproxyTransparentIfEnabled() (*mitmTransparent, error) {
 		ConfDir:    strings.TrimSpace(os.Getenv(constants.EnvMitmproxyConfDir)),
 		ScriptPath: strings.TrimSpace(os.Getenv(constants.EnvMitmproxyScript)),
 	}
-	restartCh := make(chan error, 1)
-	cfg.OnExit = func(err error) {
-		select {
-		case restartCh <- err:
-		default:
-		}
-	}
-	running, err := mitmproxy.Launch(cfg)
+	// Buffer absorbs OnExit events from a retry storm so OnExit goroutines
+	// don't all park waiting for the watcher to drain. Correctness does not
+	// depend on the size: launchTagged uses a blocking send with shutdownCh
+	// as the only escape, so events cannot be silently dropped while the
+	// watcher is alive.
+	restartCh := make(chan exitEvent, 64)
+	shutdownCh := make(chan struct{})
+	const initialGen uint64 = 1
+	running, err := launchTagged(cfg, restartCh, shutdownCh, initialGen)
 	if err != nil {
 		return nil, fmt.Errorf("start mitmdump: %w", err)
 	}
@@ -93,44 +135,111 @@ func startMitmproxyTransparentIfEnabled() (*mitmTransparent, error) {
 	if err := mitmproxy.SyncRootCA(confDir, mpHome); err != nil {
 		return nil, fmt.Errorf("mitm CA export: %w", err)
 	}
-	return &mitmTransparent{running: running, port: mpPort, uid: mpUID, cfg: cfg, restartCh: restartCh}, nil
+	return &mitmTransparent{
+		running:    running,
+		currentGen: initialGen,
+		port:       mpPort,
+		uid:        mpUID,
+		cfg:        cfg,
+		nextGen:    initialGen,
+		restartCh:  restartCh,
+		shutdownCh: shutdownCh,
+	}, nil
 }
 
 // watchMitmproxy monitors mitmdump for unexpected exits, logs the error, and restarts it.
 // Must be called after startMitmproxyTransparentIfEnabled.
 func (m *mitmTransparent) watchMitmproxy(ctx context.Context, gate *mitmproxy.HealthGate) {
+	// Closing shutdownCh on ctx cancel unblocks any OnExit closures that are
+	// parked on the (now-unread) restartCh send so they don't leak past
+	// shutdown.
+	safego.Go(func() {
+		<-ctx.Done()
+		close(m.shutdownCh)
+	})
 	safego.Go(func() {
 		for {
 			select {
-			case err := <-m.restartCh:
+			case ev := <-m.restartCh:
 				select {
 				case <-ctx.Done():
 					return
 				default:
 				}
+				cur := m.getCurrentGen()
+				if ev.gen != cur {
+					// Stale event: a previous half-launched attempt that we
+					// killed is just now being reaped. The currently-live
+					// mitmdump is unaffected; ignore and keep watching.
+					log.Infof("[mitmproxy] ignoring stale exit event (gen=%d, current=%d): %v", ev.gen, cur, ev.err)
+					continue
+				}
 
-				log.Errorf("[mitmproxy] mitmdump exited: %v; restarting...", err)
+				log.Errorf("[mitmproxy] mitmdump exited (gen=%d): %v; restarting...", ev.gen, ev.err)
 				gate.SetReady(false)
+				m.restartWithBackoff(ctx, gate)
 
-				newRunning, launchErr := mitmproxy.Launch(m.cfg)
-				if launchErr != nil {
-					log.Errorf("[mitmproxy] failed to restart mitmdump: %v; giving up", launchErr)
-					return
-				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	})
+}
 
-				waitAddr := fmt.Sprintf("127.0.0.1:%d", m.cfg.ListenPort)
-				if waitErr := mitmproxy.WaitListenPort(waitAddr, 15*time.Second); waitErr != nil {
-					log.Errorf("[mitmproxy] restart: wait listen %s: %v; giving up", waitAddr, waitErr)
-					return
-				}
+// restartWithBackoff retries mitmdump launch indefinitely with exponential backoff
+// (1s, 2s, 4s, ..., capped at 30s) until it succeeds or ctx is cancelled.
+// Transient OOM / resource pressure must not leave egress in a permanent dead state.
+//
+// Each attempt is tagged with a fresh generation; setRunning publishes that
+// generation as the "live" one. Exit events for older (killed) generations are
+// filtered out by watchMitmproxy, so we do NOT drain restartCh here -- doing
+// so could swallow a real death of the freshly-restarted mitmdump.
+func (m *mitmTransparent) restartWithBackoff(ctx context.Context, gate *mitmproxy.HealthGate) {
+	const (
+		initialBackoff = time.Second
+		maxBackoff     = 30 * time.Second
+	)
+	backoff := initialBackoff
+	waitAddr := fmt.Sprintf("127.0.0.1:%d", m.cfg.ListenPort)
+
+	for attempt := 1; ; attempt++ {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
 
-				m.setRunning(newRunning)
+		gen := atomic.AddUint64(&m.nextGen, 1)
+		newRunning, launchErr := launchTagged(m.cfg, m.restartCh, m.shutdownCh, gen)
+		if launchErr == nil {
+			if waitErr := mitmproxy.WaitListenPort(waitAddr, 15*time.Second); waitErr == nil {
+				m.setRunning(newRunning, gen)
 				gate.SetReady(true)
-				log.Infof("[mitmproxy] mitmdump restarted (pid %d)", newRunning.Cmd.Process.Pid)
-
-			case <-ctx.Done():
+				log.Infof("[mitmproxy] mitmdump restarted (pid %d, gen %d, attempt %d)", newRunning.Cmd.Process.Pid, gen, attempt)
 				return
+			} else {
+				log.Errorf("[mitmproxy] restart attempt %d (gen %d): wait listen %s: %v", attempt, gen, waitAddr, waitErr)
+				// GracefulShutdown SIGTERMs then SIGKILLs and waits for reap, so
+				// the listen port is released before the next attempt's Launch
+				// races to bind it. Direct Process.Kill returns immediately and
+				// can cause spurious WaitListenPort failures on port contention.
+				mitmproxy.GracefulShutdown(newRunning, time.Second)
 			}
+		} else {
+			log.Errorf("[mitmproxy] restart attempt %d (gen %d): launch failed: %v", attempt, gen, launchErr)
 		}
-	})
+
+		log.Warnf("[mitmproxy] restart attempt %d failed; retrying in %s", attempt, backoff)
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(backoff):
+		}
+		if backoff < maxBackoff {
+			backoff *= 2
+			if backoff > maxBackoff {
+				backoff = maxBackoff
+			}
+		}
+	}
 }