test(watch): run the identity-change monitor test on synctest's virtual clock (TASK-4067) (#2316)

* test(watch): run the identity-change monitor test on synctest's virtual clock (TASK-4067) TestMonitorChatIdentityChangeResetsMemory drove runWatchMonitor over a real loopback httptest.Server and paid real reconnect and identity backoffs (32s). It now runs in a synctest bubble over an in-memory net.Pipe listener, with http.DefaultTransport swapped for the bubble's life, and asserts exact virtual elapsed time per case. Every existing assertion is unchanged; the handler's 1ms ack poll is replaced by a channel so the times are exact. TestMonitorChatServerFloorHunterLocalDelivered pins both sides of the first-contact floor max(server floor, local delivered). Claude-Session: https://claude.ai/code/session_01Hk4vxsaKDm3E1LvsoHtHeV * test(watch): state and enforce the DefaultTransport swap's no-parallel precondition (TASK-4067 r1) Claude-Session: https://claude.ai/code/session_01Hk4vxsaKDm3E1LvsoHtHeV
2026-11-01 17:09:55 +00:00 · 2026-10-25 16:28:57 -04:00
parent 6402eb3055
commit ccb4ed4a22
3 changed files with 203 additions and 16 deletions
@@ -54,6 +54,40 @@ func TestMonitorChatServerFloorHunter(t *testing.T) {
 	}
 }

+// T1 with local delivery: the first-contact floor is max(server floor, local
+// delivered) (cmd_watch_chat.go attemptFloor). Each row pins one side: below
+// the floor, local must not reopen rows the server says are delivered; above
+// it, the server floor must not re-acknowledge what this process already did
+// (TASK-4067). The event is the same out-of-order tombstone, seq 3.
+func TestMonitorChatServerFloorHunterLocalDelivered(t *testing.T) {
+	for _, row := range []struct {
+		name         string
+		local        int64
+		wantAcks     []int64
+		wantFloor    int64
+		wantDelivery int64
+	}{
+		// Floor 2: seq 3 is the next contiguous tombstone, consumed directly.
+		{name: "local-below-floor", local: 1, wantAcks: []int64{3}, wantFloor: 3, wantDelivery: 3},
+		// Floor 3: seq 3 is already delivered, so nothing is acknowledged.
+		{name: "local-above-floor", local: 3, wantAcks: nil, wantFloor: 3, wantDelivery: 3},
+	} {
+		t.Run(row.name, func(t *testing.T) {
+			chatSessionEnv(t)
+			f := &fakeChatSink{floor: 2, messages: []monitorChatMessage{chatMsg("m1", 1), chatMsg("m2", 2)}}
+			c := newMonitorChat(f, newRecentSet(256, 10*time.Minute))
+			c.delivered[2] = row.local
+			var out bytes.Buffer
+			p := floorEvent(2, 3)
+			p.Deleted = true
+			c.handle(&out, p)
+			if f.floorReads.Load() != 1 || c.last[2] != row.wantFloor || c.delivered[2] != row.wantDelivery || !slices.Equal(f.ackSeqs, row.wantAcks) || f.fetches != 0 || out.Len() != 0 {
+				t.Fatalf("floor reads=%d last=%v delivered=%v acks=%v fetches=%d after=%v output=%q", f.floorReads.Load(), c.last, c.delivered, f.ackSeqs, f.fetches, f.after, out.String())
+			}
+		})
+	}
+}
+
 // B2: a lone DM must be delivered during this call, with no second event.
 func TestMonitorChatFirstContactDeliversSameEvent(t *testing.T) {
 	chatSessionEnv(t)
@@ -7,7 +7,6 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
-	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"slices"
@@ -107,16 +106,25 @@ func TestMonitorChatIdentityChangeResetsMemory(t *testing.T) {
 			oldSeen := seenChat
 			seenChat = newRecentSet(256, 10*time.Minute)
 			t.Cleanup(func() { seenChat = oldSeen })
-			ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
-			defer cancel()
+			// Assigned inside the bubble by runMonitorInBubble's setup.
+			var ctx context.Context
+			var cancel context.CancelFunc
 			var mu sync.Mutex
 			connections, profileReads := 0, 0
+			// Closed and replaced on every acknowledgement, so the stream
+			// handler waits for one without polling (exact virtual time).
+			// Made in setup: a channel made outside the bubble never blocks
+			// durably, and the bubble's clock would stop on it.
+			var ackSignal chan struct{}
+			// Virtual time of the first acknowledgement, from the monitor's start.
+			var bubbleStart time.Time
+			firstAck := time.Duration(-1)
 			var afters, acks []int64
 			var serverURL string
 			writeURL := func(base string) error {
 				return os.WriteFile(filepath.Join(".", ".pad.toml"), []byte(fmt.Sprintf("workspace = \"test\"\nurl = %q\n", base)), 0600)
 			}
-			h := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 				mu.Lock()
 				defer mu.Unlock()
 				path := strings.TrimPrefix(strings.TrimPrefix(r.URL.Path, "/first"), "/second")
@@ -165,14 +173,14 @@ func TestMonitorChatIdentityChangeResetsMemory(t *testing.T) {
 					// Hold the stream until this single event has been acknowledged.
 					for {
 						mu.Lock()
-						handled := len(acks) >= connections
+						handled, acked := len(acks) >= connections, ackSignal
 						mu.Unlock()
 						if handled {
 							break
 						}
 						select {
 						case <-ctx.Done():
-						case <-time.After(time.Millisecond):
+						case <-acked:
 							continue
 						}
 						break
@@ -213,6 +221,11 @@ func TestMonitorChatIdentityChangeResetsMemory(t *testing.T) {
 						t.Error(err)
 					}
 					acks = append(acks, body.Seq)
+					if firstAck < 0 {
+						firstAck = time.Since(bubbleStart)
+					}
+					close(ackSignal)
+					ackSignal = make(chan struct{})
 					if connections == 1 && change == "base-url" {
 						if err := writeURL(serverURL + "/second"); err != nil {
 							t.Error(err)
@@ -226,19 +239,48 @@ func TestMonitorChatIdentityChangeResetsMemory(t *testing.T) {
 					t.Error("unexpected request", r.URL)
 					http.NotFound(w, r)
 				}
-			}))
-			defer h.Close()
-			serverURL = h.URL
-			if err := writeURL(serverURL + "/first"); err != nil {
-				t.Fatal(err)
-			}
-			out := captureStdoutDraining(t, func() {
-				if err := runWatchMonitor(ctx); err != nil {
-					t.Error(err)
-				}
 			})
+			out, elapsed, err := runMonitorInBubble(t, 25*time.Second, handler, func(bubbleCtx context.Context, bubbleCancel context.CancelFunc, url string) error {
+				ctx, cancel, serverURL = bubbleCtx, bubbleCancel, url
+				ackSignal = make(chan struct{})
+				bubbleStart = time.Now()
+				return writeURL(serverURL + "/first")
+			})
+			if err != nil {
+				t.Error(err)
+			}
 			mu.Lock()
 			defer mu.Unlock()
+			// TASK-4067: exact virtual time. One reconnect backoff (padddBackoff(1),
+			// 5s) between the two connections; profile-once also waits out the
+			// identity backoff (monitorChatFloorBackoff, 10s) plus 1s before its
+			// handler re-sends; the unresolved cases end at the handler's own
+			// 500ms cancel, with no reconnect. Literals, not the constants, so a
+			// change to either backoff turns this red rather than moving with it.
+			wantElapsed := 5 * time.Second
+			switch change {
+			case "profile-error", "empty-user":
+				wantElapsed = 500 * time.Millisecond
+			case "profile-once":
+				wantElapsed = 16 * time.Second
+			}
+			if elapsed != wantElapsed {
+				t.Fatalf("identity %s: virtual elapsed %v, want %v", change, elapsed, wantElapsed)
+			}
+			// The first connection's event is acknowledged at once, except in
+			// profile-once: its failed connect lookup backs identity off globally,
+			// so only the handler's re-send at 11s (10s backoff + 1s) is handled.
+			// The unresolved cases never acknowledge.
+			wantFirstAck := time.Duration(0)
+			switch change {
+			case "profile-error", "empty-user":
+				wantFirstAck = -1
+			case "profile-once":
+				wantFirstAck = 11 * time.Second
+			}
+			if firstAck != wantFirstAck {
+				t.Fatalf("identity %s: first acknowledgement at virtual %v, want %v", change, firstAck, wantFirstAck)
+			}
 			if change == "profile-error" || change == "empty-user" {
 				if connections != 1 || len(afters) != 0 || len(acks) != 0 || !strings.Contains(out, "push-survives") || strings.Contains(out, "fetched ") {
 					t.Fatalf("unresolved identity affected push or handled chat: connections=%d after=%v acks=%v output=%q", connections, afters, acks, out)
@@ -0,0 +1,111 @@
+package main
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"os/signal"
+	"sync"
+	"testing"
+	"testing/synctest"
+	"time"
+)
+
+// memNet is an in-memory listener for a synctest bubble. Loopback TCP
+// sockets do not block durably, so a bubble's clock never advances past an
+// httptest.Server; net.Pipe does block durably (TASK-4067).
+type memNet struct {
+	conns  chan net.Conn
+	closed chan struct{}
+	once   sync.Once
+}
+
+func newMemNet() *memNet {
+	return &memNet{conns: make(chan net.Conn), closed: make(chan struct{})}
+}
+
+func (m *memNet) Accept() (net.Conn, error) {
+	select {
+	case c := <-m.conns:
+		return c, nil
+	case <-m.closed:
+		return nil, net.ErrClosed
+	}
+}
+
+func (m *memNet) Close() error {
+	m.once.Do(func() { close(m.closed) })
+	return nil
+}
+
+func (m *memNet) Addr() net.Addr { return &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 80} }
+
+func (m *memNet) dial(ctx context.Context, _, _ string) (net.Conn, error) {
+	client, server := net.Pipe()
+	select {
+	case m.conns <- server:
+		return client, nil
+	case <-m.closed:
+		return nil, net.ErrClosed
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+// runMonitorInBubble runs runWatchMonitor end to end in a synctest bubble,
+// against handler served over a memNet, so its reconnect and identity
+// backoffs cost virtual time only. setup runs inside the bubble before the
+// monitor starts, with the bubble's ctx (bounded by timeout) and the
+// server's URL. elapsed is the monitor's virtual run time.
+//
+// Every client on the monitor's path uses http.DefaultTransport (the stream's
+// &http.Client{} and cli.newClientFromURL's clients), so the bubble swaps it
+// for a memNet transport and restores it in Cleanup. A process-global swap is
+// safe only from a sequential top-level test with no parallel subtests (a
+// caller that uses t.Setenv, as the monitor tests do, cannot be parallel). No
+// other test can run during it: a top-level t.Parallel test is resumed only
+// after every sequential top-level test has returned, and a parallel subtest
+// finishes inside its own parent, which returns before the next top-level
+// test starts. The helper enforces the precondition with t.Setenv.
+func runMonitorInBubble(t *testing.T, timeout time.Duration, handler http.Handler, setup func(ctx context.Context, cancel context.CancelFunc, serverURL string) error) (out string, elapsed time.Duration, runErr error) {
+	t.Helper()
+	// Enforces the no-parallel precondition above: t.Setenv panics in a
+	// parallel test (or one with a parallel ancestor).
+	t.Setenv("PAD_TEST_DEFAULT_TRANSPORT_SWAPPED", "1")
+	// os/signal starts its process-wide watcher goroutine on the first
+	// Notify. Started inside a bubble, that goroutine would belong to it
+	// and block forever in a receive that is not durable, so the bubble
+	// could never finish. Start it here, outside.
+	_, stopSignals := signal.NotifyContext(context.Background(), os.Interrupt)
+	stopSignals()
+	// captureStdoutDraining's reader blocks on an os.Pipe, which is not
+	// durable either, so the bubble runs inside the capture, not around it.
+	out = captureStdoutDraining(t, func() {
+		synctest.Test(t, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), timeout)
+			defer cancel()
+			mn := newMemNet()
+			h := httptest.NewUnstartedServer(handler)
+			h.Listener = mn
+			h.Start()
+			oldTransport := http.DefaultTransport
+			transport := &http.Transport{DialContext: mn.dial}
+			http.DefaultTransport = transport
+			// Cleanup runs inside the bubble, before synctest.Test returns.
+			t.Cleanup(func() {
+				http.DefaultTransport = oldTransport
+				transport.CloseIdleConnections()
+				h.Close()
+			})
+			if err := setup(ctx, cancel, h.URL); err != nil {
+				t.Fatal(err)
+			}
+			start := time.Now()
+			runErr = runWatchMonitor(ctx)
+			elapsed = time.Since(start)
+		})
+	})
+	return out, elapsed, runErr
+}