Merge pull request 'fix(control): ack a stop before delivering it so a closing owner cannot drop the reply' (#6) from fix/control-stop-ack into main

Reviewed-on: https://git.bnema.dev/bnema/nefercap/pulls/6

commit: 1d850a025d164ed0e2f20b6f989b03abb7f4ebd3
author: bnema · 10d ago · 2026-10-12 13:18 UTC
merge of: 9e4b5ff + ba459e9 · diff against the first parent
signature: SSH UnverifiedSHA256:VOCA5lYARHcbKDZ7GhTV45VmcjWnJ6ZQIBazj17LivA The key is unknown

2 files · +120 -9

M internal/adapters/control/control.go +41 -9


@@ -367,22 +367,20 @@ func (s Server) serve(c net.UnixConn) {
367	367		if !ok {
368	368		return
369	369		}
370		-	var reply string
371	370		switch {
372	371		case bytes.Equal(line, []byte(stopLine)):
373		-	// Coalescing, non-blocking: the ack means the stop was requested,
374		-	// not that the capture file is finalized.
	372	+	// Ack before delivering: the owner may Close, and so drop this
	373	+	// connection, as soon as it sees the request. The ack means the stop
	374	+	// was requested, not that the capture file is finalized.
	375	+	io.WriteString(c, okLine)
	376	+	// Coalescing, non-blocking.
375	377		select {
376	378		case s.reqs <- RequestStop:
377	379		default:
378	380		}
379		-	reply = okLine
380	381		case bytes.Equal(line, []byte(statusLine)):
381		-	reply = recLine
382		-	default:
383		-	return
	382	+	io.WriteString(c, recLine)
384	383		}
385		-	io.WriteString(c, reply)
386	384		}
387	385
388	386		// readLine reads one newline-terminated request of at most maxRequest bytes.

@@ -486,11 +484,15 @@ func exchange(ctx context.Context, runtime, req, want string) (bool, error) {
486	484		} else if uid != euid {
487	485		return false, fmt.Errorf("%w: socket served by another user", ErrInsecure)
488	486		}
	487	+	dev, ino := uint64(st.Dev), uint64(st.Ino)
489	488		if _, err := io.WriteString(uc, req); err != nil {
490		-	return false, ctxErr(ctx, err)
	489	+	return false, droppedErr(ctx, dirFD, dev, ino, err)
491	490		}
492	491		var buf [len(recLine)]byte // the longest reply
493	492		n, err := io.ReadFull(uc, buf[:len(want)])
	493	+	if n == 0 && err != nil {
	494	+	return false, droppedErr(ctx, dirFD, dev, ino, err)
	495	+	}
494	496		if err != nil {
495	497		return false, ctxErr(ctx, err)
496	498		}

@@ -519,6 +521,36 @@ func dialRetry(ctx context.Context, path string) (net.Conn, error) {
519	521		}
520	522		}
521	523
	524	+	// droppedErr handles a connection closed before any reply. Close drops open
	525	+	// connections, then removes the socket, so an owner shutting down is seen
	526	+	// as the socket inode going away: that is no recording, not an error. A
	527	+	// server that dropped us and keeps serving yields the original error once
	528	+	// ctx ends; a cancelled caller gets context.Canceled.
	529	+	func droppedErr(ctx context.Context, dirFD int, dev, ino uint64, err error) error {
	530	+	if !errors.Is(err, io.EOF) && !errors.Is(err, unix.ECONNRESET) && !errors.Is(err, unix.EPIPE) {
	531	+	return ctxErr(ctx, err)
	532	+	}
	533	+	for !socketGone(dirFD, dev, ino) {
	534	+	t := time.NewTimer(5 * time.Millisecond)
	535	+	select {
	536	+	case <-t.C:
	537	+	case <-ctx.Done():
	538	+	t.Stop()
	539	+	if errors.Is(ctx.Err(), context.Canceled) {
	540	+	return ctx.Err()
	541	+	}
	542	+	return fmt.Errorf("control: server closed the connection: %w", err)
	543	+	}
	544	+	}
	545	+	return nil
	546	+	}
	547	+
	548	+	func socketGone(dirFD int, dev, ino uint64) bool {
	549	+	var st unix.Stat_t
	550	+	err := unix.Fstatat(dirFD, sockName, &st, unix.AT_SYMLINK_NOFOLLOW)
	551	+	return err == unix.ENOENT \|\| err == nil && (uint64(st.Dev) != dev \|\| uint64(st.Ino) != ino)
	552	+	}
	553	+
522	554		func ctxErr(ctx context.Context, err error) error {
523	555		if cerr := ctx.Err(); cerr != nil {
524	556		return cerr

 	if !ok {
 		return
 	}
-	var reply string
 	switch {
 	case bytes.Equal(line, []byte(stopLine)):
-		// Coalescing, non-blocking: the ack means the stop was requested,
+		// Ack before delivering: the owner may Close, and so drop this
-		// not that the capture file is finalized.
+		// connection, as soon as it sees the request. The ack means the stop
+		// was requested, not that the capture file is finalized.
+		io.WriteString(c, okLine)
+		// Coalescing, non-blocking.
 		select {
 		case s.reqs <- RequestStop:
 		default:
 		}
-		reply = okLine
 	case bytes.Equal(line, []byte(statusLine)):
-		reply = recLine
+		io.WriteString(c, recLine)
-	default:
-		return
 	}
-	io.WriteString(c, reply)
 }
 // readLine reads one newline-terminated request of at most maxRequest bytes.
 	} else if uid != euid {
 		return false, fmt.Errorf("%w: socket served by another user", ErrInsecure)
 	}
+	dev, ino := uint64(st.Dev), uint64(st.Ino)
 	if _, err := io.WriteString(uc, req); err != nil {
-		return false, ctxErr(ctx, err)
+		return false, droppedErr(ctx, dirFD, dev, ino, err)
 	}
 	var buf [len(recLine)]byte // the longest reply
 	n, err := io.ReadFull(uc, buf[:len(want)])
+	if n == 0 && err != nil {
+		return false, droppedErr(ctx, dirFD, dev, ino, err)
+	}
 	if err != nil {
 		return false, ctxErr(ctx, err)
 	}
 	}
 }
+// droppedErr handles a connection closed before any reply. Close drops open
+// connections, then removes the socket, so an owner shutting down is seen
+// as the socket inode going away: that is no recording, not an error. A
+// server that dropped us and keeps serving yields the original error once
+// ctx ends; a cancelled caller gets context.Canceled.
+func droppedErr(ctx context.Context, dirFD int, dev, ino uint64, err error) error {
+	if !errors.Is(err, io.EOF) && !errors.Is(err, unix.ECONNRESET) && !errors.Is(err, unix.EPIPE) {
+		return ctxErr(ctx, err)
+	}
+	for !socketGone(dirFD, dev, ino) {
+		t := time.NewTimer(5 * time.Millisecond)
+		select {
+		case <-t.C:
+		case <-ctx.Done():
+			t.Stop()
+			if errors.Is(ctx.Err(), context.Canceled) {
+				return ctx.Err()
+			}
+			return fmt.Errorf("control: server closed the connection: %w", err)
+		}
+	}
+	return nil
+}
+func socketGone(dirFD int, dev, ino uint64) bool {
+	var st unix.Stat_t
+	err := unix.Fstatat(dirFD, sockName, &st, unix.AT_SYMLINK_NOFOLLOW)
+	return err == unix.ENOENT || err == nil && (uint64(st.Dev) != dev || uint64(st.Ino) != ino)
+}
 func ctxErr(ctx context.Context, err error) error {
 	if cerr := ctx.Err(); cerr != nil {
 		return cerr

M internal/adapters/control/control_test.go +79 -0


@@ -531,6 +531,85 @@ func TestStopAckDoesNotBlockOnUnreadRequests(t *testing.T) {
531	531		}
532	532		}
533	533
	534	+	// The owner may Close as soon as it sees the stop; the client must still get
	535	+	// its ack instead of EOF. Timing-dependent: the old order failed here only
	536	+	// under load, so this guards the race rather than proving the order.
	537	+	func TestStopAckSurvivesImmediateClose(t *testing.T) {
	538	+	setup(t)
	539	+	c := ctx(t)
	540	+	for i := 0; i < 200; i++ {
	541	+	s := listenT(t)
	542	+	closed := make(chan struct{})
	543	+	go func() {
	544	+	defer close(closed)
	545	+	select {
	546	+	case <-s.Requests():
	547	+	s.Close()
	548	+	case <-c.Done():
	549	+	}
	550	+	}()
	551	+	ok, err := Stop(c)
	552	+	require.NoError(t, err, "attempt %d", i)
	553	+	require.True(t, ok)
	554	+	select {
	555	+	case <-closed:
	556	+	case <-time.After(5 * time.Second):
	557	+	t.Fatal("close did not finish")
	558	+	}
	559	+	}
	560	+	}
	561	+
	562	+	// Clients racing the owner's shutdown see either an ack or no recording,
	563	+	// never an error.
	564	+	func TestStopDuringShutdownIsNotAnError(t *testing.T) {
	565	+	setup(t)
	566	+	c := ctx(t)
	567	+	for i := 0; i < 50; i++ {
	568	+	s := listenT(t)
	569	+	go func() {
	570	+	select {
	571	+	case <-s.Requests():
	572	+	s.Close()
	573	+	case <-c.Done():
	574	+	}
	575	+	}()
	576	+	var wg sync.WaitGroup
	577	+	errs := make(chan error, 8)
	578	+	for j := 0; j < cap(errs); j++ {
	579	+	wg.Add(1)
	580	+	go func() {
	581	+	defer wg.Done()
	582	+	_, err := Stop(c)
	583	+	errs <- err
	584	+	}()
	585	+	}
	586	+	wg.Wait()
	587	+	close(errs)
	588	+	for err := range errs {
	589	+	require.NoError(t, err, "attempt %d", i)
	590	+	}
	591	+	require.NoError(t, s.Close())
	592	+	}
	593	+	}
	594	+
	595	+	// A server that drops the client but keeps serving is an error, and a
	596	+	// caller cancelled while the client waits gets context.Canceled.
	597	+	func TestDroppedByLiveServer(t *testing.T) {
	598	+	setup(t)
	599	+	s, err := listen(os.Getenv("XDG_RUNTIME_DIR"), uint32(os.Geteuid())+1)
	600	+	require.NoError(t, err)
	601	+	t.Cleanup(func() { s.Close() })
	602	+
	603	+	_, err = Stop(ctx(t))
	604	+	require.Error(t, err)
	605	+	assert.Contains(t, err.Error(), "control: server closed the connection")
	606	+
	607	+	c, cancel := context.WithCancel(context.Background())
	608	+	time.AfterFunc(100*time.Millisecond, cancel)
	609	+	_, err = Stop(c)
	610	+	assert.ErrorIs(t, err, context.Canceled)
	611	+	}
	612	+
534	613		func TestCloseLifecycle(t *testing.T) {
535	614		rd := setup(t)
536	615		s, err := Listen()


@@ -531,6 +531,85 @@ func TestStopAckDoesNotBlockOnUnreadRequests(t *testing.T) {
531	}	531		}
532	}	532		}
533		533
		534	+	// The owner may Close as soon as it sees the stop; the client must still get
		535	+	// its ack instead of EOF. Timing-dependent: the old order failed here only
		536	+	// under load, so this guards the race rather than proving the order.
		537	+	func TestStopAckSurvivesImmediateClose(t *testing.T) {
		538	+	setup(t)
		539	+	c := ctx(t)
		540	+	for i := 0; i < 200; i++ {
		541	+	s := listenT(t)
		542	+	closed := make(chan struct{})
		543	+	go func() {
		544	+	defer close(closed)
		545	+	select {
		546	+	case <-s.Requests():
		547	+	s.Close()
		548	+	case <-c.Done():
		549	+	}
		550	+	}()
		551	+	ok, err := Stop(c)
		552	+	require.NoError(t, err, "attempt %d", i)
		553	+	require.True(t, ok)
		554	+	select {
		555	+	case <-closed:
		556	+	case <-time.After(5 * time.Second):
		557	+	t.Fatal("close did not finish")
		558	+	}
		559	+	}
		560	+	}
		561	+
		562	+	// Clients racing the owner's shutdown see either an ack or no recording,
		563	+	// never an error.
		564	+	func TestStopDuringShutdownIsNotAnError(t *testing.T) {
		565	+	setup(t)
		566	+	c := ctx(t)
		567	+	for i := 0; i < 50; i++ {
		568	+	s := listenT(t)
		569	+	go func() {
		570	+	select {
		571	+	case <-s.Requests():
		572	+	s.Close()
		573	+	case <-c.Done():
		574	+	}
		575	+	}()
		576	+	var wg sync.WaitGroup
		577	+	errs := make(chan error, 8)
		578	+	for j := 0; j < cap(errs); j++ {
		579	+	wg.Add(1)
		580	+	go func() {
		581	+	defer wg.Done()
		582	+	_, err := Stop(c)
		583	+	errs <- err
		584	+	}()
		585	+	}
		586	+	wg.Wait()
		587	+	close(errs)
		588	+	for err := range errs {
		589	+	require.NoError(t, err, "attempt %d", i)
		590	+	}
		591	+	require.NoError(t, s.Close())
		592	+	}
		593	+	}
		594	+
		595	+	// A server that drops the client but keeps serving is an error, and a
		596	+	// caller cancelled while the client waits gets context.Canceled.
		597	+	func TestDroppedByLiveServer(t *testing.T) {
		598	+	setup(t)
		599	+	s, err := listen(os.Getenv("XDG_RUNTIME_DIR"), uint32(os.Geteuid())+1)
		600	+	require.NoError(t, err)
		601	+	t.Cleanup(func() { s.Close() })
		602	+
		603	+	_, err = Stop(ctx(t))
		604	+	require.Error(t, err)
		605	+	assert.Contains(t, err.Error(), "control: server closed the connection")
		606	+
		607	+	c, cancel := context.WithCancel(context.Background())
		608	+	time.AfterFunc(100*time.Millisecond, cancel)
		609	+	_, err = Stop(c)
		610	+	assert.ErrorIs(t, err, context.Canceled)
		611	+	}
		612	+
534	func TestCloseLifecycle(t *testing.T) {	613		func TestCloseLifecycle(t *testing.T) {
535	rd := setup(t)	614		rd := setup(t)
536	s, err := Listen()	615		s, err := Listen()