fix(control): report no recording when the owner closes during a request
- commit
- ba459e952020012203604991c3be16233f522c13
- author
- brice · · 2026-10-12 05:37 UTC
- parent
- 3d02f43
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +86 -1
M internal/adapters/control/control.go +35 -1
| @@ -484,11 +484,15 @@ func exchange(ctx context.Context, runtime, req, want string) (bool, error) { | |||
| 484 | 484 | } else if uid != euid { | |
| 485 | 485 | return false, fmt.Errorf("%w: socket served by another user", ErrInsecure) | |
| 486 | 486 | } | |
| 487 | + | dev, ino := uint64(st.Dev), uint64(st.Ino) | |
| 487 | 488 | if _, err := io.WriteString(uc, req); err != nil { | |
| 488 | - | return false, ctxErr(ctx, err) | |
| 489 | + | return false, droppedErr(ctx, dirFD, dev, ino, err) | |
| 489 | 490 | } | |
| 490 | 491 | var buf [len(recLine)]byte // the longest reply | |
| 491 | 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 | + | } | |
| 492 | 496 | if err != nil { | |
| 493 | 497 | return false, ctxErr(ctx, err) | |
| 494 | 498 | } | |
| @@ -517,6 +521,36 @@ func dialRetry(ctx context.Context, path string) (net.Conn, error) { | |||
| 517 | 521 | } | |
| 518 | 522 | } | |
| 519 | 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 | + | ||
| 520 | 554 | func ctxErr(ctx context.Context, err error) error { | |
| 521 | 555 | if cerr := ctx.Err(); cerr != nil { | |
| 522 | 556 | return cerr | |
| @@ -484,11 +484,15 @@ func exchange(ctx context.Context, runtime, req, want string) (bool, error) { | |||||
| 484 | } else if uid != euid { | 484 | } else if uid != euid { | ||
| 485 | return false, fmt.Errorf("%w: socket served by another user", ErrInsecure) | 485 | return false, fmt.Errorf("%w: socket served by another user", ErrInsecure) | ||
| 486 | } | 486 | } | ||
| 487 | + | dev, ino := uint64(st.Dev), uint64(st.Ino) | |||
| 487 | if _, err := io.WriteString(uc, req); err != nil { | 488 | if _, err := io.WriteString(uc, req); err != nil { | ||
| 488 | - | return false, ctxErr(ctx, err) | 489 | + | return false, droppedErr(ctx, dirFD, dev, ino, err) |
| 489 | } | 490 | } | ||
| 490 | var buf [len(recLine)]byte // the longest reply | 491 | var buf [len(recLine)]byte // the longest reply | ||
| 491 | n, err := io.ReadFull(uc, buf[:len(want)]) | 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 | + | } | |||
| 492 | if err != nil { | 496 | if err != nil { | ||
| 493 | return false, ctxErr(ctx, err) | 497 | return false, ctxErr(ctx, err) | ||
| 494 | } | 498 | } | ||
| @@ -517,6 +521,36 @@ func dialRetry(ctx context.Context, path string) (net.Conn, error) { | |||||
| 517 | } | 521 | } | ||
| 518 | } | 522 | } | ||
| 519 | 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 | + | ||||
| 520 | func ctxErr(ctx context.Context, err error) error { | 554 | func ctxErr(ctx context.Context, err error) error { | ||
| 521 | if cerr := ctx.Err(); cerr != nil { | 555 | if cerr := ctx.Err(); cerr != nil { | ||
| 522 | return cerr | 556 | return cerr | ||
M internal/adapters/control/control_test.go +51 -0
| @@ -559,6 +559,57 @@ func TestStopAckSurvivesImmediateClose(t *testing.T) { | |||
| 559 | 559 | } | |
| 560 | 560 | } | |
| 561 | 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 | + | ||
| 562 | 613 | func TestCloseLifecycle(t *testing.T) { | |
| 563 | 614 | rd := setup(t) | |
| 564 | 615 | s, err := Listen() | |
| @@ -559,6 +559,57 @@ func TestStopAckSurvivesImmediateClose(t *testing.T) { | |||||
| 559 | } | 559 | } | ||
| 560 | } | 560 | } | ||
| 561 | 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 | + | ||||
| 562 | func TestCloseLifecycle(t *testing.T) { | 613 | func TestCloseLifecycle(t *testing.T) { | ||
| 563 | rd := setup(t) | 614 | rd := setup(t) | ||
| 564 | s, err := Listen() | 615 | s, err := Listen() | ||