Maven/internal/ipc/cancel_test.go

package ipc

import (
	"context"
	"errors"
	"net"
	"path/filepath"
	"testing"
	"time"
)

// A cancelled context has to abort a call that is already in flight. It did not
// until V-638: call checked ctx once before sending and then blocked in
// roundtrip with no connection deadline, so a daemon that read the frame and
// never answered parked the caller for as long as the socket stayed open.
//
// The server here is that daemon: it accepts, reads nothing, replies nothing.

func deafServer(t *testing.T) string {
	t.Helper()
	sock := filepath.Join(t.TempDir(), "deaf.sock")
	ln, err := net.Listen("unix", sock)
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	t.Cleanup(func() { _ = ln.Close() })
	go func() {
		for {
			conn, err := ln.Accept()
			if err != nil {
				return
			}
			// Hold it open and say nothing. Closed by the listener cleanup.
			t.Cleanup(func() { _ = conn.Close() })
		}
	}()
	return sock
}

func TestClientCancelAbortsAReadInFlight(t *testing.T) {
	c, err := Dial(deafServer(t))
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	defer c.Close()

	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		time.Sleep(50 * time.Millisecond)
		cancel()
	}()

	done := make(chan error, 1)
	go func() {
		_, err := c.Ping(ctx)
		done <- err
	}()

	select {
	case err := <-done:
		// Ping is read-only, so the cancellation is reported as itself rather
		// than as an ambiguous mutation.
		if !errors.Is(err, context.Canceled) {
			t.Errorf("got %v, want context.Canceled", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("a cancelled Ping did not return")
	}
}

// A mutation cancelled while awaiting the reply may already have committed, so
// it is ErrAmbiguousOutcome and never a retry. That split is the invariant
// internal/ipc/maperr_test.go's neighbours rest on.
func TestClientCancelLeavesAMutationAmbiguous(t *testing.T) {
	c, err := Dial(deafServer(t))
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	defer c.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	defer cancel()

	done := make(chan error, 1)
	go func() {
		_, err := c.WriteFact(ctx, WriteFactReq{Key: "water", Value: "drank"})
		done <- err
	}()

	select {
	case err := <-done:
		if !errors.Is(err, ErrAmbiguousOutcome) {
			t.Errorf("got %v, want ErrAmbiguousOutcome", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("a cancelled WriteFact did not return")
	}
}

// The deadline itself, with no cancellation: a call on a context with no
// deadline used to have no bound at all. This one has one and must respect it.
func TestClientDeadlineBoundsACall(t *testing.T) {
	c, err := Dial(deafServer(t))
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	defer c.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer cancel()

	start := time.Now()
	if _, err := c.Ping(ctx); err == nil {
		t.Fatal("a deaf server answered a Ping")
	}
	if elapsed := time.Since(start); elapsed > 3*time.Second {
		t.Errorf("Ping took %v, want the context deadline to bound it", elapsed)
	}
}

// blockingAPI parks Presence until its context is cancelled and records what
// cancelled it. Every other method is the unimplemented floor.
type blockingAPI struct {
	UnimplementedCoreAPI
	entered chan struct{}
	err     chan error
}

func (b *blockingAPI) Presence(ctx context.Context) (Presence, error) {
	close(b.entered)
	<-ctx.Done()
	b.err <- ctx.Err()
	return Presence{}, ctx.Err()
}

// serveConn dispatched under context.Background() until V-638, so Close could
// only abandon a dispatch in flight and never tell it to stop.
func TestServerCloseCancelsADispatchInFlight(t *testing.T) {
	api := &blockingAPI{entered: make(chan struct{}), err: make(chan error, 1)}
	srv, err := Listen(filepath.Join(t.TempDir(), "core.sock"), api)
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	served := make(chan struct{})
	go func() { _ = srv.Serve(); close(served) }()

	cli, err := Dial(srv.Path())
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	defer cli.Close()
	go func() { _, _ = cli.Presence(context.Background()) }()

	select {
	case <-api.entered:
	case <-time.After(5 * time.Second):
		t.Fatal("the handler was never dispatched")
	}

	_ = srv.Close()
	<-served
	select {
	case got := <-api.err:
		if !errors.Is(got, context.Canceled) {
			t.Errorf("handler saw %v, want context.Canceled", got)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("Close did not cancel the dispatch")
	}
}

// The watchdog closes the conn, and it races the end of the call: a
// cancellation landing as the reply arrives can close a conn the call was
// already done with. That is survivable either way, because a write to a closed
// socket is errWriteLost and errWriteLost re-dials and retries, so this test
// passes with or without the drop in roundtrip's defer. What it pins is that
// the recovery is real and costs one round trip at most, never an error the
// caller sees.
func TestClientSurvivesACancelledCall(t *testing.T) {
	_, _, cli, _ := newServerWithStore(t)

	for i := 0; i < 20; i++ {
		ctx, cancel := context.WithCancel(context.Background())
		go cancel() // races the reply on purpose
		_, _ = cli.Ping(ctx)
		cancel()

		if _, err := cli.Ping(context.Background()); err != nil {
			t.Fatalf("call %d after a cancelled one: %v", i, err)
		}
	}
}