fix(control): ack a stop before delivering it so a closing owner cannot drop the reply
- commit
- 3d02f434cf07f322c5af08b39d7e0e3cab663f0e
- author
- brice · · 2026-10-12 05:28 UTC
- parent
- c00bc5f
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +34 -8
M internal/adapters/control/control.go +6 -8
| @@ -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. | |
| @@ -367,22 +367,20 @@ func (s *Server) serve(c *net.UnixConn) { | |||||
| 367 | if !ok { | 367 | if !ok { | ||
| 368 | return | 368 | return | ||
| 369 | } | 369 | } | ||
| 370 | - | var reply string | |||
| 371 | switch { | 370 | switch { | ||
| 372 | case bytes.Equal(line, []byte(stopLine)): | 371 | case bytes.Equal(line, []byte(stopLine)): | ||
| 373 | - | // Coalescing, non-blocking: the ack means the stop was requested, | 372 | + | // Ack before delivering: the owner may Close, and so drop this |
| 374 | - | // not that the capture file is finalized. | 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 | select { | 377 | select { | ||
| 376 | case s.reqs <- RequestStop: | 378 | case s.reqs <- RequestStop: | ||
| 377 | default: | 379 | default: | ||
| 378 | } | 380 | } | ||
| 379 | - | reply = okLine | |||
| 380 | case bytes.Equal(line, []byte(statusLine)): | 381 | case bytes.Equal(line, []byte(statusLine)): | ||
| 381 | - | reply = recLine | 382 | + | io.WriteString(c, recLine) |
| 382 | - | default: | |||
| 383 | - | return | |||
| 384 | } | 383 | } | ||
| 385 | - | io.WriteString(c, reply) | |||
| 386 | } | 384 | } | ||
| 387 | 385 | ||||
| 388 | // readLine reads one newline-terminated request of at most maxRequest bytes. | 386 | // readLine reads one newline-terminated request of at most maxRequest bytes. | ||
M internal/adapters/control/control_test.go +28 -0
| @@ -531,6 +531,34 @@ 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 | + | ||
| 534 | 562 | func TestCloseLifecycle(t *testing.T) { | |
| 535 | 563 | rd := setup(t) | |
| 536 | 564 | s, err := Listen() | |
| @@ -531,6 +531,34 @@ 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 | + | ||||
| 534 | func TestCloseLifecycle(t *testing.T) { | 562 | func TestCloseLifecycle(t *testing.T) { | ||
| 535 | rd := setup(t) | 563 | rd := setup(t) | ||
| 536 | s, err := Listen() | 564 | s, err := Listen() | ||