client: fix a data race on a witness's old size

internal/witness/client.go

@@ -297,6 +297,11 @@ type witnessClient struct {
 	client wc.Witness
 	// verifiers are verifiers for the signature(s) returned by the witness.
 	verifiers []note.Verifier
+
+	// mu is held for the whole of an update, and guards oldSize.
+	// An update may still be running when the next one starts: CosignCheckpoint returns cosignatures
+	// as they arrive, and the caller can stop reading them once its policy is met.
+	mu sync.Mutex
 	// oldSize is the size of the checkpoint that the log thinks that the witness last signed.
 	// This can be zero, in which case the tlog-witness protocol discovery mechanism is used to determine the old size.
 	oldSize uint64
@@ -314,6 +319,12 @@ func names(w []note.Verifier) []string {
 }
 
 func (w *witnessClient) update(ctx context.Context, cp []byte, cpSize uint64, fetchProof func(ctx context.Context, from, to uint64) ([][]byte, error)) ([]byte, error) {
+	// Updates to a witness run one at a time, so that each starts from the size the last one left.
+	// If an older update gets the lock after a newer one, its fetchProof fails, as oldSize is then
+	// larger than its checkpoint, so it can't set oldSize back.
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
 	const maxUpdateRetries = 3
 	// Set InitialInterval low so that we handle catching up after StatusConflict/ErrCheckpointStale quickly.
 	// But ramp up quickly to avoid DoSing witnesses.

internal/witness/client_test.go

@@ -475,6 +475,171 @@ func TestWitnessReusesProofs(t *testing.T) {
 	}
 }
 
+// TestWitnessGatewayUnreadUpdates checks, under -race, that the gateway can send out the next
+// checkpoint while a witness's update for the previous one hasn't been read. A log whose policy
+// is met by either of two witnesses stops reading at the first cosignature.
+func TestWitnessGatewayUnreadUpdates(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		_, checkpoint, ok := bytes.Cut(body, []byte("\n\n"))
+		if !ok {
+			t.Errorf("expected two newlines in body, got: %q", body)
+			return
+		}
+		_, _, n, err := log.ParseCheckpoint(checkpoint, logVerifier.Name(), logVerifier)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		switch r.URL.Path {
+		case "/wit1/add-checkpoint":
+			_, _ = w.Write(sigForSigner(t, n.Text, wit1Skey))
+		case "/wit2/add-checkpoint":
+			_, _ = w.Write(sigForSigner(t, n.Text, wit2Skey))
+		default:
+			t.Errorf("Unknown case: %s", r.URL.String())
+		}
+	}))
+	defer ts.Close()
+	baseURL := mustURL(t, ts.URL)
+	wit1Verifier, err := f_note.NewVerifierForCosignatureV1(wit1Vkey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	wit2Verifier, err := f_note.NewVerifierForCosignatureV1(wit2Vkey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	g, err := witness.NewGateway(t.Context(), witness.Options{
+		HTTPClient: ts.Client(),
+		Witnesses: []witness.Witness{
+			{URL: baseURL.JoinPath("wit1"), Verifiers: []note.Verifier{wit1Verifier}},
+			{URL: baseURL.JoinPath("wit2"), Verifiers: []note.Verifier{wit2Verifier}},
+		},
+		FetchTiles: testLogTileFetcher,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var unread []<-chan []byte
+	for size := 1; size < 10; size++ {
+		logSignedCheckpoint, _ := loadCheckpoint(t, size)
+		sigs := g.CosignCheckpoint(t.Context(), logSignedCheckpoint, uint64(size))
+		if _, ok := <-sigs; !ok {
+			t.Fatalf("size %d: got no signatures", size)
+		}
+		unread = append(unread, sigs)
+		// Give the other witness's update time to finish, as a log's checkpoint interval does.
+		time.Sleep(10 * time.Millisecond)
+	}
+	for i, sigs := range unread {
+		if _, n := collectSigs(t, sigs); n != 1 {
+			t.Errorf("size %d: got %d more signatures, want 1", i+1, n)
+		}
+	}
+}
+
+func TestWitnessGatewayLateUpdate(t *testing.T) {
+	// The old and checkpoint sizes of each request that wit2 receives, in order.
+	type request struct{ old, size uint64 }
+	var (
+		mu       sync.Mutex
+		requests []request
+	)
+	held := make(chan struct{})
+	release := make(chan struct{})
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		head, checkpoint, ok := bytes.Cut(body, []byte("\n\n"))
+		if !ok {
+			t.Errorf("expected two newlines in body, got: %q", body)
+			return
+		}
+		var old uint64
+		if _, err := fmt.Sscanf(string(head), "old %d", &old); err != nil {
+			t.Errorf("parsing old size from %q: %v", head, err)
+			return
+		}
+		cp, _, n, err := log.ParseCheckpoint(checkpoint, logVerifier.Name(), logVerifier)
+		if err != nil {
+			t.Error(err)
+			return
+		}
+		switch r.URL.Path {
+		case "/wit1/add-checkpoint":
+			_, _ = w.Write(sigForSigner(t, n.Text, wit1Skey))
+		case "/wit2/add-checkpoint":
+			mu.Lock()
+			requests = append(requests, request{old: old, size: cp.Size})
+			mu.Unlock()
+			// Hold wit2's first update until the next checkpoint has gone out.
+			if cp.Size == 1 {
+				close(held)
+				<-release
+			}
+			_, _ = w.Write(sigForSigner(t, n.Text, wit2Skey))
+		default:
+			t.Errorf("Unknown case: %s", r.URL.String())
+		}
+	}))
+	defer ts.Close()
+	baseURL := mustURL(t, ts.URL)
+	wit1Verifier, err := f_note.NewVerifierForCosignatureV1(wit1Vkey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	wit2Verifier, err := f_note.NewVerifierForCosignatureV1(wit2Vkey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	g, err := witness.NewGateway(t.Context(), witness.Options{
+		HTTPClient: ts.Client(),
+		Witnesses: []witness.Witness{
+			{URL: baseURL.JoinPath("wit1"), Verifiers: []note.Verifier{wit1Verifier}},
+			{URL: baseURL.JoinPath("wit2"), Verifiers: []note.Verifier{wit2Verifier}},
+		},
+		FetchTiles: testLogTileFetcher,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cp1, _ := loadCheckpoint(t, 1)
+	cp2, _ := loadCheckpoint(t, 2)
+	cp3, _ := loadCheckpoint(t, 3)
+
+	// Send checkpoint 2 while wit2's update for checkpoint 1 is still running, so that update finishes late.
+	sigs1 := g.CosignCheckpoint(t.Context(), cp1, 1)
+	<-held
+	sigs2 := g.CosignCheckpoint(t.Context(), cp2, 2)
+	time.Sleep(10 * time.Millisecond)
+	close(release)
+	for i, sigs := range []<-chan []byte{sigs1, sigs2} {
+		if _, n := collectSigs(t, sigs); n != 2 {
+			t.Errorf("size %d: got %d signatures, want 2", i+1, n)
+		}
+	}
+	if _, n := collectSigs(t, g.CosignCheckpoint(t.Context(), cp3, 3)); n != 2 {
+		t.Errorf("size 3: got %d signatures, want 2", n)
+	}
+
+	// Each update must start from the size the last one left, and the late update mustn't set it back.
+	want := []request{{old: 0, size: 1}, {old: 1, size: 2}, {old: 2, size: 3}}
+	mu.Lock()
+	defer mu.Unlock()
+	if !slices.Equal(requests, want) {
+		t.Errorf("wit2 got requests (old, size) %v, want %v", requests, want)
+	}
+}
+
 func loadCheckpoint(t *testing.T, size int) (signed []byte, unsigned string) {
 	t.Helper()
 	path := fmt.Sprintf("../../testdata/log/checkpoint.%d", size)