From 6e216dd5f1e3563dd638bc0ce44af7dc2c5fc5ef Mon Sep 17 00:00:00 2001 From: Glenn Rempe Date: Thu, 5 Nov 2026 19:26:10 -0500 Subject: [PATCH] witness: fix a data race on a witness's old size CosignCheckpoint returns cosignatures as they arrive, and the log stops reading them once its policy is met. When either of two witnesses can meet the policy, the other's update can still be running, or finish unread, when the next checkpoint goes out. That update stores the witness's oldSize, and the next update reads it, with nothing ordering the two, so the race detector reports a data race. An update that finishes late can also set oldSize back to an older size. Give each witness a mutex, held for the whole update, so that its updates run one at a time, each starting from the size the last one left. --- internal/witness/client.go | 11 +++ internal/witness/client_test.go | 165 ++++++++++++++++++++++++++++++++ 2 files changed, 176 insertions(+) diff --git a/internal/witness/client.go b/internal/witness/client.go index bc9b67450..882027511 100644 --- a/internal/witness/client.go +++ b/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. diff --git a/internal/witness/client_test.go b/internal/witness/client_test.go index e46733baf..a7555e018 100644 --- a/internal/witness/client_test.go +++ b/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)