From db2d3485d4f99112b2be5bf15a69e2913dc31559 Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Fri, 31 Jul 2026 16:10:57 +0000 Subject: [PATCH 1/5] Mirror client fixes --- client/mirror/client.go | 13 +- client/mirror/client_test.go | 269 +++++++++++++++++++++++------------ 2 files changed, 188 insertions(+), 94 deletions(-) diff --git a/client/mirror/client.go b/client/mirror/client.go index 657690b79..76a3e4433 100644 --- a/client/mirror/client.go +++ b/client/mirror/client.go @@ -315,6 +315,10 @@ func (c *Client) streamEntries(ctx context.Context, uploadStart, uploadEnd uint6 startIdx := curr % 256 endIdx := startIdx + numEntries + if uint64(len(bundle.Entries)) < endIdx { + _ = pw.CloseWithError(fmt.Errorf("bundle %d has only %d entries, expected at least %d", bundleIndex, len(bundle.Entries), endIdx)) + return + } for i := startIdx; i < endIdx; i++ { entry := bundle.Entries[i] if err := binary.Write(gw, binary.BigEndian, uint16(len(entry))); err != nil { @@ -443,10 +447,14 @@ func (c *Client) buildCheckpointRequestBody(oldSize uint64, proof [][]byte, chec // up to the specified targetSize. It returns the mirror's cosignatures on success. func (c *Client) Sync(ctx context.Context, targetCheckpointRaw []byte, targetSize uint64) ([]byte, error) { var conflict ErrConflict - nextEntry := c.oldSize + var nextEntry uint64 // Push the checkpoint with the old size (0 if not provided). - for c.oldSize < targetSize { + // Keep trying for as long we we get conflict errors or the context is not cancelled. + // Ensure we send the checkpoint at least once, this serves two purposes: + // 1. We refresh the witness' timestamped view of the log, committing to the fact that the log hasn't grown. + // 2. If this is the first time we're pushing the checkpoint (i.e. c.oldSize == 0), we're ensuring that we'll send a zero-sized checkpoint. + for { err := c.pushCheckpoint(ctx, c.oldSize, targetSize, targetCheckpointRaw) if err != nil { if !errors.As(err, &conflict) { @@ -458,6 +466,7 @@ func (c *Client) Sync(ctx context.Context, targetCheckpointRaw []byte, targetSiz nextEntry = c.oldSize c.oldSize = targetSize + break } // Push entries up to target size in packages of 256, handling concurrent conflicts and retries. diff --git a/client/mirror/client_test.go b/client/mirror/client_test.go index 59096bf1c..40bd99804 100644 --- a/client/mirror/client_test.go +++ b/client/mirror/client_test.go @@ -403,6 +403,7 @@ type fakeMirror struct { initialStatus int addEntriesExpectations []addEntriesExpectation addEntriesCallCount int + expectClientError bool } type addEntriesExpectation struct { @@ -428,6 +429,13 @@ func newFakeMirror(t *testing.T, origin string) *fakeMirror { } } +func (fm *fakeMirror) handleReadError(err error, msg string, w http.ResponseWriter) { + if !fm.expectClientError { + fm.t.Errorf("%s: %v", msg, err) + } + w.WriteHeader(http.StatusBadRequest) +} + // ServeHTTP handles incoming HTTP requests to mock mirror endpoints (/add-entries, /add-checkpoint). func (fm *fakeMirror) ServeHTTP(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { @@ -439,8 +447,7 @@ func (fm *fakeMirror) ServeHTTP(w http.ResponseWriter, r *http.Request) { } gr, err := gzip.NewReader(r.Body) if err != nil { - fm.t.Errorf("gzip.NewReader failed: %v", err) - w.WriteHeader(http.StatusBadRequest) + fm.handleReadError(err, "gzip.NewReader failed", w) return } defer func() { @@ -449,14 +456,12 @@ func (fm *fakeMirror) ServeHTTP(w http.ResponseWriter, r *http.Request) { var originLen uint16 if err := binary.Read(gr, binary.BigEndian, &originLen); err != nil { - fm.t.Errorf("failed to read origin length: %v", err) - w.WriteHeader(http.StatusBadRequest) + fm.handleReadError(err, "failed to read origin length", w) return } gotOrigin := make([]byte, originLen) if _, err := io.ReadFull(gr, gotOrigin); err != nil { - fm.t.Errorf("failed to read origin: %v", err) - w.WriteHeader(http.StatusBadRequest) + fm.handleReadError(err, "failed to read origin", w) return } if string(gotOrigin) != fm.origin { @@ -465,26 +470,22 @@ func (fm *fakeMirror) ServeHTTP(w http.ResponseWriter, r *http.Request) { var start uint64 if err := binary.Read(gr, binary.BigEndian, &start); err != nil { - fm.t.Errorf("failed to read start: %v", err) - w.WriteHeader(http.StatusBadRequest) + fm.handleReadError(err, "failed to read start", w) return } var end uint64 if err := binary.Read(gr, binary.BigEndian, &end); err != nil { - fm.t.Errorf("failed to read end: %v", err) - w.WriteHeader(http.StatusBadRequest) + fm.handleReadError(err, "failed to read end", w) return } var gotTicketLen uint16 if err := binary.Read(gr, binary.BigEndian, &gotTicketLen); err != nil { - fm.t.Errorf("failed to read ticket length: %v", err) - w.WriteHeader(http.StatusBadRequest) + fm.handleReadError(err, "failed to read ticket length", w) return } gotTicket := make([]byte, gotTicketLen) if _, err := io.ReadFull(gr, gotTicket); err != nil { - fm.t.Errorf("failed to read ticket: %v", err) - w.WriteHeader(http.StatusBadRequest) + fm.handleReadError(err, "failed to read ticket", w) return } @@ -539,14 +540,12 @@ func (fm *fakeMirror) ServeHTTP(w http.ResponseWriter, r *http.Request) { for i := range numExpectedEntries { var entryLen uint16 if err := binary.Read(gr, binary.BigEndian, &entryLen); err != nil { - fm.t.Errorf("failed to read entry %d length: %v", i, err) - w.WriteHeader(http.StatusBadRequest) + fm.handleReadError(err, fmt.Sprintf("failed to read entry %d length", i), w) return } entry := make([]byte, entryLen) if _, err := io.ReadFull(gr, entry); err != nil { - fm.t.Errorf("failed to read entry %d: %v", i, err) - w.WriteHeader(http.StatusBadRequest) + fm.handleReadError(err, fmt.Sprintf("failed to read entry %d", i), w) return } expectedEntry := fmt.Appendf(nil, "entry-%d", i+int(start)) @@ -557,29 +556,29 @@ func (fm *fakeMirror) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } - var numHashes uint8 - if err := binary.Read(gr, binary.BigEndian, &numHashes); err != nil { - fm.t.Errorf("failed to read numHashes: %v", err) - w.WriteHeader(http.StatusBadRequest) - return - } - if int(numHashes) != len(fm.proofHashes) { - fm.t.Errorf("numHashes = %d, want %d", numHashes, len(fm.proofHashes)) - w.WriteHeader(http.StatusBadRequest) - return - } - for i, expectedHash := range fm.proofHashes { - gotHash := make([]byte, 32) - if _, err := io.ReadFull(gr, gotHash); err != nil { - fm.t.Errorf("failed to read hash %d: %v", i, err) - w.WriteHeader(http.StatusBadRequest) + if start < end { + var numHashes uint8 + if err := binary.Read(gr, binary.BigEndian, &numHashes); err != nil { + fm.handleReadError(err, "failed to read numHashes", w) return } - if !bytes.Equal(gotHash, expectedHash) { - fm.t.Errorf("hash %d = %x, want %x", i, gotHash, expectedHash) + if int(numHashes) != len(fm.proofHashes) { + fm.t.Errorf("numHashes = %d, want %d", numHashes, len(fm.proofHashes)) w.WriteHeader(http.StatusBadRequest) return } + for i, expectedHash := range fm.proofHashes { + gotHash := make([]byte, 32) + if _, err := io.ReadFull(gr, gotHash); err != nil { + fm.handleReadError(err, fmt.Sprintf("failed to read hash %d", i), w) + return + } + if !bytes.Equal(gotHash, expectedHash) { + fm.t.Errorf("hash %d = %x, want %x", i, gotHash, expectedHash) + w.WriteHeader(http.StatusBadRequest) + return + } + } } // Consume the rest of the gzipped body (including the gzip footer) to prevent client write hangs. @@ -638,7 +637,7 @@ func (fm *fakeMirror) ServeHTTP(w http.ResponseWriter, r *http.Request) { _, _ = fmt.Fprintf(w, "%d\n", fm.initialPendingSize) return } - if oldSize > 0 && len(headerLines) <= 1 { + if oldSize > 0 && oldSize < uint64(fm.numEntries) && len(headerLines) <= 1 { fm.t.Errorf("expected consistency proof lines in header") w.WriteHeader(http.StatusBadRequest) return @@ -665,65 +664,130 @@ func (fm *fakeMirror) ServeHTTP(w http.ResponseWriter, r *http.Request) { // checkpoint updating, and entry streaming. func TestSync(t *testing.T) { origin := "test-origin" - fm := newFakeMirror(t, origin) - fm.addEntriesExpectations = []addEntriesExpectation{ + + type syncStep struct { + checkpoint []byte + targetSize uint64 + beforeSync func(fm *fakeMirror) + expectCosigs []byte + } + + for _, tc := range []struct { + desc string + addEntriesExpectations []addEntriesExpectation + steps []syncStep + }{ { - start: 0, - end: uint64(fm.numEntries), - ticket: nil, - status: http.StatusOK, + desc: "single sync", + addEntriesExpectations: []addEntriesExpectation{ + { + start: 0, + end: 5, + ticket: nil, + status: http.StatusOK, + }, + }, + steps: []syncStep{ + { + checkpoint: []byte("checkpoint-raw-bytes"), + targetSize: 5, + expectCosigs: []byte("mock-cosignatures"), + }, + }, }, - } - server := httptest.NewServer(fm) - defer server.Close() + { + desc: "sync twice", + addEntriesExpectations: []addEntriesExpectation{ + { + start: 0, + end: 5, + ticket: nil, + status: http.StatusOK, + }, + { + start: 5, + end: 5, + ticket: nil, + status: http.StatusOK, + }, + }, + steps: []syncStep{ + { + checkpoint: []byte("checkpoint-raw-bytes"), + targetSize: 5, + expectCosigs: []byte("mock-cosignatures"), + }, + { + checkpoint: []byte("checkpoint-raw-bytes"), + targetSize: 5, + expectCosigs: []byte("mock-cosignatures"), + beforeSync: func(fm *fakeMirror) { + fm.initialPendingSize = 5 + }, + }, + }, + }, + } { + t.Run(tc.desc, func(t *testing.T) { + fm := newFakeMirror(t, origin) + fm.addEntriesExpectations = tc.addEntriesExpectations - u, err := url.Parse(server.URL + "/") - if err != nil { - t.Fatalf("failed to parse server URL: %v", err) - } + server := httptest.NewServer(fm) + defer server.Close() - tileFetcher := func(ctx context.Context, level, index uint64, p uint8) ([]byte, error) { - return nil, errors.New("tile fetcher should not be called") - } + u, err := url.Parse(server.URL + "/") + if err != nil { + t.Fatalf("failed to parse server URL: %v", err) + } - bundleFetcher := func(ctx context.Context, bundleIndex uint64, p uint8) ([]byte, error) { - var buf bytes.Buffer - for i := range fm.numEntries { - entry := fmt.Appendf(nil, "entry-%d", i) - _ = binary.Write(&buf, binary.BigEndian, uint16(len(entry))) - buf.Write(entry) - } - return buf.Bytes(), nil - } + tileFetcher := func(ctx context.Context, level, index uint64, p uint8) ([]byte, error) { + return nil, errors.New("tile fetcher should not be called") + } - mirrorCheckpointFetcher := func(ctx context.Context) ([]byte, error) { - return nil, errors.New("mirror checkpoint fetcher should not be called") - } + bundleFetcher := func(ctx context.Context, bundleIndex uint64, p uint8) ([]byte, error) { + var buf bytes.Buffer + for i := range fm.numEntries { + entry := fmt.Appendf(nil, "entry-%d", i) + _ = binary.Write(&buf, binary.BigEndian, uint16(len(entry))) + buf.Write(entry) + } + return buf.Bytes(), nil + } - packageProver := func(ctx context.Context, start, end, size uint64) ([][]byte, error) { - return fm.proofHashes, nil - } + mirrorCheckpointFetcher := func(ctx context.Context) ([]byte, error) { + return nil, errors.New("mirror checkpoint fetcher should not be called") + } - opts := NewOptions(). - WithMirrorURL(u). - WithLogOrigin(origin). - WithTileFetcher(tileFetcher). - WithBundleFetcher(bundleFetcher). - WithMirrorCheckpointFetcher(mirrorCheckpointFetcher). - WithPackageProver(packageProver) + packageProver := func(ctx context.Context, start, end, size uint64) ([][]byte, error) { + return fm.proofHashes, nil + } - c, err := NewClient(context.Background(), opts) - if err != nil { - t.Fatalf("NewClient() = _, %v, want _, nil", err) - } + opts := NewOptions(). + WithMirrorURL(u). + WithLogOrigin(origin). + WithTileFetcher(tileFetcher). + WithBundleFetcher(bundleFetcher). + WithMirrorCheckpointFetcher(mirrorCheckpointFetcher). + WithPackageProver(packageProver) - cosigs, err := c.Sync(context.Background(), fm.targetCheckpoint, uint64(fm.numEntries)) - if err != nil { - t.Fatalf("Sync() = _, %v, want _, nil", err) - } + c, err := NewClient(context.Background(), opts) + if err != nil { + t.Fatalf("NewClient() = _, %v, want _, nil", err) + } - if !bytes.Equal(cosigs, fm.expectedCosigs) { - t.Errorf("Sync() = %q, want %q", string(cosigs), string(fm.expectedCosigs)) + for i, step := range tc.steps { + if step.beforeSync != nil { + step.beforeSync(fm) + } + cosigs, err := c.Sync(context.Background(), step.checkpoint, step.targetSize) + if err != nil { + t.Fatalf("Sync() step %d = _, %v, want _, nil", i+1, err) + } + if !bytes.Equal(cosigs, step.expectCosigs) { + t.Errorf("Sync() step %d = %q, want %q", i+1, string(cosigs), string(step.expectCosigs)) + } + } + }) } } @@ -738,6 +802,7 @@ func TestSync_ErrorsAndEdgeCases(t *testing.T) { addEntriesExpectations []addEntriesExpectation addCheckpointStatus int tileFetcher client.TileFetcherFunc + bundleFetcher client.EntryBundleFetcherFunc wantErr string }{ { @@ -842,6 +907,22 @@ func TestSync_ErrorsAndEdgeCases(t *testing.T) { }, wantErr: "mirror size reverted to 2, which is smaller than target 5", }, + { + desc: "bundle has fewer entries than expected", + initialPendingSize: 0, + initialNextEntry: 0, + bundleFetcher: func(ctx context.Context, bundleIndex uint64, p uint8) ([]byte, error) { + var buf bytes.Buffer + // Return only 2 entries even though fm.numEntries is 5 + for i := range 2 { + entry := fmt.Appendf(nil, "entry-%d", i) + _ = binary.Write(&buf, binary.BigEndian, uint16(len(entry))) + buf.Write(entry) + } + return buf.Bytes(), nil + }, + wantErr: "bundle 0 has only 2 entries, expected at least 5", + }, } for _, tc := range tests { @@ -854,6 +935,7 @@ func TestSync_ErrorsAndEdgeCases(t *testing.T) { if tc.addCheckpointStatus != 0 { fm.addCheckpointStatus = tc.addCheckpointStatus } + fm.expectClientError = tc.wantErr != "" server := httptest.NewServer(fm) defer server.Close() @@ -870,14 +952,17 @@ func TestSync_ErrorsAndEdgeCases(t *testing.T) { } } - bundleFetcher := func(ctx context.Context, bundleIndex uint64, p uint8) ([]byte, error) { - var buf bytes.Buffer - for i := range fm.numEntries { - entry := fmt.Appendf(nil, "entry-%d", i) - _ = binary.Write(&buf, binary.BigEndian, uint16(len(entry))) - buf.Write(entry) + bundleFetcher := tc.bundleFetcher + if bundleFetcher == nil { + bundleFetcher = func(ctx context.Context, bundleIndex uint64, p uint8) ([]byte, error) { + var buf bytes.Buffer + for i := range fm.numEntries { + entry := fmt.Appendf(nil, "entry-%d", i) + _ = binary.Write(&buf, binary.BigEndian, uint16(len(entry))) + buf.Write(entry) + } + return buf.Bytes(), nil } - return buf.Bytes(), nil } mirrorCheckpointFetcher := func(ctx context.Context) ([]byte, error) { From c2cb4115eb3642927cf72e98aee639cd3edf49d6 Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Mon, 27 Jul 2026 11:35:21 +0000 Subject: [PATCH 2/5] Add mirror gateway. --- append_lifecycle.go | 53 +++- append_lifecycle_test.go | 167 +++++++++++- internal/mirror/gateway/mirror_gateway.go | 254 ++++++++++++++++++ .../mirror/gateway/mirror_gateway_test.go | 181 +++++++++++++ storage/aws/aws.go | 2 +- storage/gcp/gcp.go | 2 +- storage/posix/files.go | 2 +- 7 files changed, 638 insertions(+), 23 deletions(-) create mode 100644 internal/mirror/gateway/mirror_gateway.go create mode 100644 internal/mirror/gateway/mirror_gateway_test.go diff --git a/append_lifecycle.go b/append_lifecycle.go index 0169c3c28..8ab1f512c 100644 --- a/append_lifecycle.go +++ b/append_lifecycle.go @@ -15,6 +15,7 @@ package tessera import ( + "bytes" "context" "errors" "fmt" @@ -30,6 +31,7 @@ import ( f_log "github.com/transparency-dev/formats/log" "github.com/transparency-dev/merkle/rfc6962" "github.com/transparency-dev/tessera/api/layout" + m_gateway "github.com/transparency-dev/tessera/internal/mirror/gateway" "github.com/transparency-dev/tessera/internal/otel" "github.com/transparency-dev/tessera/internal/parse" "github.com/transparency-dev/tessera/internal/witness" @@ -730,7 +732,12 @@ func (o *AppendOptions) WithAntispam(inMemEntries uint, as Antispam) *AppendOpti } // CheckpointPublisher returns a function which should be used to create, sign, and potentially witness a new checkpoint. -func (o AppendOptions) CheckpointPublisher(lr LogReader, httpClient *http.Client) func(context.Context, uint64, []byte) ([]byte, error) { +func (o AppendOptions) CheckpointPublisher(ctx context.Context, lr LogReader, httpClient *http.Client) func(context.Context, uint64, []byte) ([]byte, error) { + var gw *m_gateway.Gateway + if len(o.mirrors.Components) > 0 { + gw = m_gateway.NewGateway(ctx, httpClient, o.mirrors, lr, o.primarySigner.Name()) + } + return func(ctx context.Context, size uint64, root []byte) ([]byte, error) { return otel.Trace(ctx, "tessera.CheckpointPublisher", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) { cp, err := o.newCP(ctx, size, root) @@ -748,8 +755,13 @@ func (o AppendOptions) CheckpointPublisher(lr LogReader, httpClient *http.Client return err }) eg.Go(func() error { + if gw == nil { + return nil + } + mirrorCtx, cancel := context.WithTimeout(ctx, o.mirrorOpts.Timeout) + defer cancel() var err error - ms, err = mirrorCheckpoint(ctx, cp, size, o.mirrors, lr, httpClient, o.mirrorOpts) + ms, err = mirrorCheckpoint(mirrorCtx, gw, &o.mirrors, cp, size, o.mirrorOpts.FailOpen) return err }) @@ -812,15 +824,38 @@ func witnessCheckpoint(ctx context.Context, cp []byte, cpSize uint64, witnesses }) } -// mirrorCheckpoint takes care of mirroring the given checkpoint with the provided mirror policy. -// Returns signatures from mirrors, ready to append to the checkpoint, or an error. -func mirrorCheckpoint(ctx context.Context, cp []byte, cpSize uint64, mirrors WitnessGroup, lr LogReader, httpClient *http.Client, opts MirroringOptions) ([]byte, error) { +func mirrorCheckpoint(ctx context.Context, gw *m_gateway.Gateway, policy *WitnessGroup, cp []byte, size uint64, failOpen bool) ([]byte, error) { + // TODO(al): Add metrics + checkPolicy := func(sigs []byte) ([]byte, error) { + newCP := append(slices.Clone(cp), sigs...) + if policy.Satisfied(newCP) { + return sigs, nil + } + if failOpen { + slog.WarnContext(ctx, "MirrorGateway: policy not met, failing-open") + return sigs, nil + } + return sigs, fmt.Errorf("MirrorGateway: policy not met") + } + return otel.Trace(ctx, "tessera.mirrorCheckpoint", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) { - if len(mirrors.Components) == 0 { - return nil, nil + sigCh := gw.CosignCheckpoint(ctx, cp, size) + var sigBlock bytes.Buffer + for { + select { + case <-ctx.Done(): + return checkPolicy(sigBlock.Bytes()) + case sig, ok := <-sigCh: + if !ok { + return checkPolicy(sigBlock.Bytes()) + } + sigBlock.Write(sig) + newCP := append(slices.Clone(cp), sigBlock.Bytes()...) + if policy.Satisfied(newCP) { + return sigBlock.Bytes(), nil + } + } } - span.AddEvent("Starting mirroring") - return nil, nil }) } diff --git a/append_lifecycle_test.go b/append_lifecycle_test.go index 128bedba6..85a4cbb32 100644 --- a/append_lifecycle_test.go +++ b/append_lifecycle_test.go @@ -15,17 +15,23 @@ package tessera import ( + "bytes" "context" + "encoding/binary" "errors" "fmt" + "io" "net/http" "net/http/httptest" "net/url" + "os" "strings" + "sync" "testing" "time" f_note "github.com/transparency-dev/formats/note" + "github.com/transparency-dev/merkle/rfc6962" "github.com/transparency-dev/witness/config" "github.com/transparency-dev/witness/persistence/inmemory" "github.com/transparency-dev/witness/witness" @@ -286,6 +292,9 @@ func TestWithMirrors(t *testing.T) { const ( testWit1VKey = "Wit1+55ee4561+AVhZSmQj9+SoL+p/nN0Hh76xXmF7QcHfytUrI1XfSClk" testWit1SKey = "PRIVATE+KEY+Wit1+55ee4561+AeadRiG7XM4XiieCHzD8lxysXMwcViy5nYsoXURWGrlE" + + testMirrorVKey = "Mirror1+66ee4561+AVhZSmQj9+SoL+p/nN0Hh76xXmF7QcHfytUrI1XfSClk" + testMirrorSKey = "PRIVATE+KEY+Mirror1+66ee4561+AeadRiG7XM4XiieCHzD8lxysXMwcViy5nYsoXURWGrlE" ) func newWitnessHandler(t *testing.T, logVerifier note.Verifier, witnessSKey string) http.HandlerFunc { @@ -347,7 +356,23 @@ func TestCheckpointPublisher(t *testing.T) { t.Fatalf("failed to create witness verifier: %v", err) } - dummyMirrors := NewWitnessGroup(1, wit) + mirrorServer := httptest.NewServer(newMirrorHandler(t, testMirrorSKey)) + t.Cleanup(mirrorServer.Close) + + mirrorServerURL, err := url.Parse(mirrorServer.URL) + if err != nil { + t.Fatalf("failed to parse mirror server url: %v", err) + } + + m, err := NewWitness(testMirrorVKey, mirrorServerURL) + if err != nil { + t.Fatalf("failed to create mirror: %v", err) + } + mirrors := NewWitnessGroup(1, m) + mirrorVerifier, err := f_note.NewVerifierForCosignatureV1(testMirrorVKey) + if err != nil { + t.Fatalf("failed to create mirror verifier: %v", err) + } for _, test := range []struct { desc string @@ -366,13 +391,14 @@ func TestCheckpointPublisher(t *testing.T) { expectCosignatures: []note.Verifier{witVerifier}, }, { - desc: "mirrors only", - opts: NewAppendOptions().WithCheckpointSigner(logSigner).WithMirrors(dummyMirrors, nil), + desc: "mirrors only", + opts: NewAppendOptions().WithCheckpointSigner(logSigner).WithMirrors(mirrors, nil), + expectCosignatures: []note.Verifier{mirrorVerifier}, }, { desc: "witnesses and mirrors", - opts: NewAppendOptions().WithCheckpointSigner(logSigner).WithWitnesses(witnesses, nil).WithMirrors(dummyMirrors, nil), - expectCosignatures: []note.Verifier{witVerifier}, + opts: NewAppendOptions().WithCheckpointSigner(logSigner).WithWitnesses(witnesses, nil).WithMirrors(mirrors, nil), + expectCosignatures: []note.Verifier{witVerifier, mirrorVerifier}, }, { desc: "witness fails, failOpen=false", @@ -403,13 +429,9 @@ func TestCheckpointPublisher(t *testing.T) { test.opts.WithWitnesses(failingWitnesses, wOpts) } - lr := &fakeLogReader{ - readCheckpoint: func(ctx context.Context) ([]byte, error) { - return nil, errors.New("no checkpoint yet") - }, - } + lr := newFakeLogReaderForTest(t) - publisher := test.opts.CheckpointPublisher(lr, client) + publisher := test.opts.CheckpointPublisher(t.Context(), lr, client) cp, err := publisher(t.Context(), 5, []byte("12345678901234567890123456789012")) if (err != nil) != test.expectErr { t.Fatalf("expected error %v but got: %v", test.expectErr, err) @@ -434,3 +456,126 @@ func TestCheckpointPublisher(t *testing.T) { }) } } + +func newFakeLogReaderForTest(t *testing.T) *fakeLogReader { + hasher := rfc6962.DefaultHasher + entries := [][]byte{ + []byte("entry-0"), + []byte("entry-1"), + []byte("entry-2"), + []byte("entry-3"), + []byte("entry-4"), + } + + h0 := hasher.HashLeaf(entries[0]) + h1 := hasher.HashLeaf(entries[1]) + h01 := hasher.HashChildren(h0, h1) + h2 := hasher.HashLeaf(entries[2]) + h3 := hasher.HashLeaf(entries[3]) + h23 := hasher.HashChildren(h2, h3) + h0123 := hasher.HashChildren(h01, h23) + h4 := hasher.HashLeaf(entries[4]) + + tileNodes := [][]byte{h0, h1, h01, h2, h3, h23, h0123, h4} + var tileBuf bytes.Buffer + for _, n := range tileNodes { + tileBuf.Write(n) + } + tileBytes := tileBuf.Bytes() + + var bundleBuf bytes.Buffer + for _, entry := range entries { + _ = binary.Write(&bundleBuf, binary.BigEndian, uint16(len(entry))) + bundleBuf.Write(entry) + } + bundleBytes := bundleBuf.Bytes() + + return &fakeLogReader{ + readCheckpoint: func(ctx context.Context) ([]byte, error) { + return nil, os.ErrNotExist + }, + readTile: func(ctx context.Context, level, index uint64, p uint8) ([]byte, error) { + if level == 0 && index == 0 { + return tileBytes, nil + } + return nil, os.ErrNotExist + }, + readEntryBundle: func(ctx context.Context, index uint64, p uint8) ([]byte, error) { + if index == 0 { + return bundleBytes, nil + } + return nil, os.ErrNotExist + }, + } +} + +func newMirrorHandler(t *testing.T, mirrorSKey string) http.HandlerFunc { + mirrorSigner, err := f_note.NewSignerForCosignatureV1(mirrorSKey) + if err != nil { + t.Fatalf("failed to create mirror signer: %v", err) + } + logVerifier, err := note.NewVerifier("example.com/log/testdata+33d7b496+AeHTu4Q3hEIMHNqc6fASMsq3rKNx280NI+oO5xCFkkSx") + if err != nil { + t.Fatalf("failed to create log verifier: %v", err) + } + + var mu sync.Mutex + var pendingCP []byte + + return func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/add-checkpoint") { + body, err := io.ReadAll(r.Body) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + parts := bytes.SplitN(body, []byte("\n\n"), 2) + if len(parts) == 2 { + mu.Lock() + pendingCP = parts[1] + mu.Unlock() + } + w.WriteHeader(http.StatusOK) + return + } + if strings.HasSuffix(r.URL.Path, "/add-entries") { + _, _ = io.Copy(io.Discard, r.Body) + + mu.Lock() + cp := pendingCP + mu.Unlock() + + if len(cp) == 0 { + w.WriteHeader(http.StatusBadRequest) + return + } + + // Open and parse the checkpoint note using log's verifier. + n, err := note.Open(cp, note.VerifierList(logVerifier)) + if err != nil { + t.Errorf("failed to open checkpoint in mock mirror: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + + // Sign it with the mirror signer. + signedNote, err := note.Sign(n, mirrorSigner) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + + // Extract only the signature line we added. + idx := strings.Index(string(signedNote), "\n— "+mirrorSigner.Name()+" ") + if idx < 0 { + w.WriteHeader(http.StatusInternalServerError) + return + } + sigLine := string(signedNote)[idx+1:] + + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(sigLine)) + return + } + } +} diff --git a/internal/mirror/gateway/mirror_gateway.go b/internal/mirror/gateway/mirror_gateway.go new file mode 100644 index 000000000..ce99f9b16 --- /dev/null +++ b/internal/mirror/gateway/mirror_gateway.go @@ -0,0 +1,254 @@ +// Copyright 2026 The Tessera authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package gateway + +import ( + "context" + "fmt" + "io" + "net/http" + "net/url" + "os" + "sync" + "time" + + "log/slog" + + "github.com/transparency-dev/tessera/client/mirror" + "golang.org/x/mod/sumdb/note" +) + +// WitnessGroup defines the subset of tessera.WitnessGroup methods needed by the gateway. +type WitnessGroup interface { + WitnessEndpoints() map[string][]note.Verifier +} + +// LogReader defines the subset of tessera.LogReader methods needed by the gateway. +type LogReader interface { + ReadTile(ctx context.Context, level, index uint64, p uint8) ([]byte, error) + ReadEntryBundle(ctx context.Context, index uint64, p uint8) ([]byte, error) +} + +// goal represents a desired state for a mirror. +// It contains a target checkpoint (and its size broken out for simplicity), and a +// callback which must be called once the goal is attained, or an error occurred. +type goal struct { + cpSize uint64 + cp []byte + done func([]byte, error) +} + +// mirrorTarget represents a tlog-mirror service which we'll attempt to update. +type mirrorTarget struct { + url *url.URL + client *mirror.Client + + goals chan goal +} + +// Gateway manages the process of keeping mirrors up-to-date. +type Gateway struct { + httpClient *http.Client + lr LogReader + targets []*mirrorTarget +} + +// NewGateway creates a new Gateway that will keep mirrors up-to-date. +func NewGateway(ctx context.Context, httpClient *http.Client, mirrors WitnessGroup, lr LogReader, logOrigin string) *Gateway { + if httpClient == nil { + httpClient = http.DefaultClient + } + + g := &Gateway{ + httpClient: httpClient, + lr: lr, + } + + endpoints := mirrors.WitnessEndpoints() + for u := range endpoints { + parsedURL, err := url.Parse(u) + if err != nil { + slog.ErrorContext(ctx, "Invalid mirror URL", slog.String("url", u), slog.Any("error", err)) + continue + } + + tileFetcher := func(ctx context.Context, level, index uint64, p uint8) ([]byte, error) { + return lr.ReadTile(ctx, level, index, p) + } + bundleFetcher := func(ctx context.Context, index uint64, p uint8) ([]byte, error) { + return lr.ReadEntryBundle(ctx, index, p) + } + mirrorCheckpointFetcher := func(ctx context.Context) ([]byte, error) { + checkpointURL, err := parsedURL.Parse("checkpoint") + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, checkpointURL.String(), nil) + if err != nil { + return nil, err + } + resp, err := httpClient.Do(req) + if err != nil { + return nil, err + } + defer func() { + _ = resp.Body.Close() + }() + if resp.StatusCode == http.StatusNotFound { + return nil, os.ErrNotExist + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("failed to fetch checkpoint from mirror: status %d", resp.StatusCode) + } + return io.ReadAll(resp.Body) + } + + mOpts := mirror.NewOptions(). + WithMirrorURL(parsedURL). + WithHTTPClient(httpClient). + WithLogOrigin(logOrigin). + WithTileFetcher(tileFetcher). + WithBundleFetcher(bundleFetcher). + WithMirrorCheckpointFetcher(mirrorCheckpointFetcher) + + c, err := mirror.NewClient(ctx, mOpts) + if err != nil { + slog.ErrorContext(ctx, "Failed to create mirror client", slog.String("url", u), slog.Any("error", err)) + continue + } + + target := &mirrorTarget{ + url: parsedURL, + client: c, + goals: make(chan goal, 1), + } + g.targets = append(g.targets, target) + + // Start the worker goroutine. + go g.runWorker(ctx, target) + } + + return g +} + +// CosignCheckpoint updates the goals for all mirrors and returns a channel on which it will send +// cosignatures as they are successfully fetched from the mirrors. +// The channel is closed once all mirrors' signatures have been sent or the context is canceled. +func (g *Gateway) CosignCheckpoint(ctx context.Context, cp []byte, cpSize uint64) <-chan []byte { + if len(g.targets) == 0 { + return nil + } + + out := make(chan []byte, len(g.targets)) + wg := sync.WaitGroup{} + + // Send goals to each of the target workers, but don't block if they're + // already busy. + for _, target := range g.targets { + newGoal := goal{ + cp: cp, + cpSize: cpSize, + done: func(sig []byte, err error) { + defer wg.Done() + if err != nil { + slog.ErrorContext(ctx, "Mirror sync failed", slog.String("url", target.url.String()), slog.Any("error", err)) + return + } + slog.InfoContext(ctx, "Mirror sync succeeded", slog.String("url", target.url.String()), slog.Uint64("size", cpSize)) + out <- sig + }, + } + + for done := false; !done; { + // Add the goal to the target worker. If there's already a goal in the channel, then we'll try to replace it since the current cosign request + // supercedes it. This is racy, but that's fine since we're only trying to replace a pending request - it's fine if the worker has already picked up + // the old goal. + select { + case <-ctx.Done(): + done = true + case target.goals <- newGoal: + // The goal was sent, we're done + wg.Add(1) + done = true + default: + // No space in the goals channel, try to supercede the goal currently in there. + select { + case oldGoal := <-target.goals: + // Ok, we've removed the superceded one, let's signal that it's done: + oldGoal.done(nil, fmt.Errorf("superseded by newer goal for size %d", cpSize)) + // Then let the loop retry the send. + default: + // Channel became empty in the meantime, let the loop retry the send. + } + } + } + } + + go func() { + wg.Wait() + close(out) + }() + + return out +} + +// runWorker runs the main loop of a mirror worker: it picks up goals from the goals channel +// and attempts to satisfy them. +// +// It will block on the goals channel until a goal is received, or the context is +// cancelled. +func (g *Gateway) runWorker(ctx context.Context, target *mirrorTarget) { + slog.InfoContext(ctx, "Starting mirror worker", slog.String("url", target.url.String())) + defer slog.InfoContext(ctx, "Stopping mirror worker", slog.String("url", target.url.String())) + + for { + select { + case <-ctx.Done(): + return + case job, ok := <-target.goals: + if !ok { + // Channel closed, stop. + return + } + + // Loop until the goal is met, or we timeout. + done := false + interval := time.Millisecond + for !done { + select { + case <-ctx.Done(): + return + case <-time.After(interval): + interval = time.Second + } + // In a func for context defer. + func() { + // TODO(al): Make this configurable? Should be plenty of time for normal operation, and we'll retry anyway if we do timeout. + cctx, cancel := context.WithTimeout(ctx, 1*time.Minute) + defer cancel() + + slog.DebugContext(cctx, "Syncing mirror", slog.String("url", target.url.String()), slog.Uint64("goal", job.cpSize)) + sigs, err := target.client.Sync(cctx, job.cp, job.cpSize) + if err != nil { + slog.WarnContext(ctx, "Mirror sync attempt failed, retrying", slog.String("url", target.url.String()), slog.Any("error", err)) + return + } + done = true + job.done(sigs, nil) + }() + } + } + } +} diff --git a/internal/mirror/gateway/mirror_gateway_test.go b/internal/mirror/gateway/mirror_gateway_test.go new file mode 100644 index 000000000..c4243d58f --- /dev/null +++ b/internal/mirror/gateway/mirror_gateway_test.go @@ -0,0 +1,181 @@ +// Copyright 2026 The Tessera authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package gateway_test + +import ( + "bytes" + "crypto/rand" + "fmt" + "io" + "net/http" + "net/http/httptest" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/transparency-dev/formats/log" + f_note "github.com/transparency-dev/formats/note" + "github.com/transparency-dev/tessera" + "github.com/transparency-dev/tessera/internal/mirror/gateway" + "github.com/transparency-dev/tessera/testonly" + "golang.org/x/mod/sumdb/note" +) + +type fakeWitnessGroup struct { + endpoints map[string][]note.Verifier +} + +func (f fakeWitnessGroup) WitnessEndpoints() map[string][]note.Verifier { + return f.endpoints +} + +func TestGateway(t *testing.T) { + opts := tessera.NewAppendOptions(). + WithCheckpointInterval(100 * time.Millisecond). + WithCheckpointRepublishInterval(100 * time.Millisecond) + + testLog, shutdown := testonly.NewTestLog(t, opts) + defer func() { + _ = shutdown(t.Context()) + }() + + // Create a log with some entries. + const size = 5 + var f tessera.IndexFuture + for i := range size { + entry := tessera.NewEntry(fmt.Appendf(nil, "entry-%d", i)) + f = testLog.Appender.Add(t.Context(), entry) + } + a := tessera.NewPublicationAwaiter(t.Context(), testLog.LogReader.ReadCheckpoint, 100*time.Millisecond) + if _, _, err := a.Await(t.Context(), f); err != nil { + t.Fatalf("failed to add entry: %v", err) + } + goalCP, err := testLog.LogReader.ReadCheckpoint(t.Context()) + if err != nil { + t.Fatalf("failed to read checkpoint: %v", err) + } + + const numMirrors = 3 + var verifiers []note.Verifier + endpoints := make(map[string][]note.Verifier) + + for i := range numMirrors { + signer, verifier := mustNewKeypair(t, fmt.Sprintf("Mirror-%d", i)) + server := startMockMirror(t, signer, testLog.SigVerifier) + defer server.Close() + + endpoints[server.URL] = []note.Verifier{verifier} + verifiers = append(verifiers, verifier) + } + + policy := fakeWitnessGroup{ + endpoints: endpoints, + } + + g := gateway.NewGateway(t.Context(), http.DefaultClient, policy, testLog.LogReader, "test") + + // Call CosignCheckpoint and gather signatures. + sigCh := g.CosignCheckpoint(t.Context(), goalCP, size) + var cosigs []byte + for sig := range sigCh { + cosigs = append(cosigs, sig...) + } + + // Verify cosignatures. + fullCP := append(slices.Clone(goalCP), cosigs...) + cp, _, n, err := log.ParseCheckpoint(fullCP, testLog.SigVerifier.Name(), testLog.SigVerifier, verifiers...) + if err != nil { + t.Fatalf("failed to verify cosigned checkpoint: %v", err) + } + if got, want := len(n.Sigs), 1+numMirrors; got != want { + t.Errorf("note signatures: got %d, want %d", got, want) + } + if got, want := uint64(cp.Size), uint64(size); got != want { + t.Errorf("checkpoint size: got %d, want %d", got, want) + } +} + +func mustNewKeypair(t *testing.T, name string) (f_note.Signer, note.Verifier) { + t.Helper() + skey, vkey, err := note.GenerateKey(rand.Reader, name) + if err != nil { + t.Fatalf("Failed to generate key: %v", err) + } + s, err := f_note.NewSignerForCosignatureV1(skey) + if err != nil { + t.Fatalf("Failed to create signer: %v", err) + } + v, err := f_note.NewVerifierForCosignatureV1(vkey) + if err != nil { + t.Fatalf("Failed to create verifier: %v", err) + } + return s, v +} + +func startMockMirror(t *testing.T, signer note.Signer, logVerifier note.Verifier) *httptest.Server { + t.Helper() + var mu sync.Mutex + var pendingCP []byte + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/add-checkpoint") { + body, _ := io.ReadAll(r.Body) + parts := bytes.SplitN(body, []byte("\n\n"), 2) + if len(parts) == 2 { + mu.Lock() + pendingCP = parts[1] + mu.Unlock() + } + w.WriteHeader(http.StatusOK) + return + } + if strings.HasSuffix(r.URL.Path, "/add-entries") { + _, _ = io.Copy(io.Discard, r.Body) + mu.Lock() + cp := pendingCP + mu.Unlock() + + if len(cp) == 0 { + w.WriteHeader(http.StatusBadRequest) + return + } + + n, err := note.Open(cp, note.VerifierList(logVerifier)) + if err != nil { + t.Errorf("failed to open cp in mock: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + signedNote, err := note.Sign(n, signer) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + return + } + idx := strings.Index(string(signedNote), "— "+signer.Name()+" ") + if idx < 0 { + w.WriteHeader(http.StatusInternalServerError) + return + } + sigLine := string(signedNote)[idx:] + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(sigLine)) + return + } + })) + + return server +} diff --git a/storage/aws/aws.go b/storage/aws/aws.go index 68e8d78b9..91a53c021 100644 --- a/storage/aws/aws.go +++ b/storage/aws/aws.go @@ -239,7 +239,7 @@ func (s *Storage) newAppender(ctx context.Context, o objStore, seq sequencer, op r := &Appender{ logStore: logStore, sequencer: seq, - newCP: opts.CheckpointPublisher(logStore, s.cfg.HTTPClient), + newCP: opts.CheckpointPublisher(ctx, logStore, s.cfg.HTTPClient), treeUpdated: make(chan struct{}), entriesAssigned: make(chan struct{}, 1), } diff --git a/storage/gcp/gcp.go b/storage/gcp/gcp.go index 3402c4c75..9c982b870 100644 --- a/storage/gcp/gcp.go +++ b/storage/gcp/gcp.go @@ -311,7 +311,7 @@ func (s *Storage) newAppender(ctx context.Context, o objStore, seq *spannerCoord }, nextIndex: a.sequencer.nextIndex, } - a.newCP = opts.CheckpointPublisher(reader, s.cfg.HTTPClient) + a.newCP = opts.CheckpointPublisher(ctx, reader, s.cfg.HTTPClient) if err := a.init(ctx); err != nil { return nil, nil, fmt.Errorf("failed to initialise log storage: %v", err) diff --git a/storage/posix/files.go b/storage/posix/files.go index a02c0c31b..c227c7572 100644 --- a/storage/posix/files.go +++ b/storage/posix/files.go @@ -151,7 +151,7 @@ func (s *Storage) newAppender(ctx context.Context, o *logResourceStorage, opts * s: s, logStorage: o, cpUpdated: make(chan struct{}), - newCP: opts.CheckpointPublisher(o, s.cfg.HTTPClient), + newCP: opts.CheckpointPublisher(ctx, o, s.cfg.HTTPClient), } if err := a.initialise(ctx); err != nil { return nil, nil, err From c1db9631f4c948f68e23ae03bbc8a0f9a9b3c365 Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Wed, 29 Jul 2026 17:14:38 +0000 Subject: [PATCH 3/5] Add retries test --- .../mirror/gateway/mirror_gateway_test.go | 146 +++++++++++------- 1 file changed, 86 insertions(+), 60 deletions(-) diff --git a/internal/mirror/gateway/mirror_gateway_test.go b/internal/mirror/gateway/mirror_gateway_test.go index c4243d58f..2de4c017b 100644 --- a/internal/mirror/gateway/mirror_gateway_test.go +++ b/internal/mirror/gateway/mirror_gateway_test.go @@ -44,68 +44,85 @@ func (f fakeWitnessGroup) WitnessEndpoints() map[string][]note.Verifier { } func TestGateway(t *testing.T) { - opts := tessera.NewAppendOptions(). - WithCheckpointInterval(100 * time.Millisecond). - WithCheckpointRepublishInterval(100 * time.Millisecond) - - testLog, shutdown := testonly.NewTestLog(t, opts) - defer func() { - _ = shutdown(t.Context()) - }() - - // Create a log with some entries. - const size = 5 - var f tessera.IndexFuture - for i := range size { - entry := tessera.NewEntry(fmt.Appendf(nil, "entry-%d", i)) - f = testLog.Appender.Add(t.Context(), entry) - } - a := tessera.NewPublicationAwaiter(t.Context(), testLog.LogReader.ReadCheckpoint, 100*time.Millisecond) - if _, _, err := a.Await(t.Context(), f); err != nil { - t.Fatalf("failed to add entry: %v", err) - } - goalCP, err := testLog.LogReader.ReadCheckpoint(t.Context()) - if err != nil { - t.Fatalf("failed to read checkpoint: %v", err) - } + for _, tc := range []struct { + name string + numMirrors int + failCount int + }{ + { + name: "multiple mirrors", + numMirrors: 3, + }, + { + name: "retry on transient error", + numMirrors: 1, + failCount: 2, + }, + } { + t.Run(tc.name, func(t *testing.T) { + opts := tessera.NewAppendOptions(). + WithCheckpointInterval(100 * time.Millisecond). + WithCheckpointRepublishInterval(100 * time.Millisecond) + + testLog, shutdown := testonly.NewTestLog(t, opts) + defer func() { + _ = shutdown(t.Context()) + }() + + // Create a log with some entries. + const size = 5 + var f tessera.IndexFuture + for i := range size { + entry := tessera.NewEntry(fmt.Appendf(nil, "entry-%d", i)) + f = testLog.Appender.Add(t.Context(), entry) + } + a := tessera.NewPublicationAwaiter(t.Context(), testLog.LogReader.ReadCheckpoint, 100*time.Millisecond) + if _, _, err := a.Await(t.Context(), f); err != nil { + t.Fatalf("failed to add entry: %v", err) + } + goalCP, err := testLog.LogReader.ReadCheckpoint(t.Context()) + if err != nil { + t.Fatalf("failed to read checkpoint: %v", err) + } - const numMirrors = 3 - var verifiers []note.Verifier - endpoints := make(map[string][]note.Verifier) + var verifiers []note.Verifier + endpoints := make(map[string][]note.Verifier) - for i := range numMirrors { - signer, verifier := mustNewKeypair(t, fmt.Sprintf("Mirror-%d", i)) - server := startMockMirror(t, signer, testLog.SigVerifier) - defer server.Close() + for i := range tc.numMirrors { + signer, verifier := mustNewKeypair(t, fmt.Sprintf("Mirror-%d", i)) + server := startMockMirror(t, signer, testLog.SigVerifier, tc.failCount) + defer server.Close() - endpoints[server.URL] = []note.Verifier{verifier} - verifiers = append(verifiers, verifier) - } + endpoints[server.URL] = []note.Verifier{verifier} + verifiers = append(verifiers, verifier) + } - policy := fakeWitnessGroup{ - endpoints: endpoints, - } + policy := fakeWitnessGroup{ + endpoints: endpoints, + } - g := gateway.NewGateway(t.Context(), http.DefaultClient, policy, testLog.LogReader, "test") + g := gateway.NewGateway(t.Context(), http.DefaultClient, policy, testLog.LogReader, "test") - // Call CosignCheckpoint and gather signatures. - sigCh := g.CosignCheckpoint(t.Context(), goalCP, size) - var cosigs []byte - for sig := range sigCh { - cosigs = append(cosigs, sig...) - } + // Call CosignCheckpoint and gather signatures. + sigCh := g.CosignCheckpoint(t.Context(), goalCP, size) + var cosigs []byte + for sig := range sigCh { + cosigs = append(cosigs, sig...) + } - // Verify cosignatures. - fullCP := append(slices.Clone(goalCP), cosigs...) - cp, _, n, err := log.ParseCheckpoint(fullCP, testLog.SigVerifier.Name(), testLog.SigVerifier, verifiers...) - if err != nil { - t.Fatalf("failed to verify cosigned checkpoint: %v", err) - } - if got, want := len(n.Sigs), 1+numMirrors; got != want { - t.Errorf("note signatures: got %d, want %d", got, want) - } - if got, want := uint64(cp.Size), uint64(size); got != want { - t.Errorf("checkpoint size: got %d, want %d", got, want) + // Verify cosignatures. + fullCP := append(slices.Clone(goalCP), cosigs...) + cp, _, n, err := log.ParseCheckpoint(fullCP, testLog.SigVerifier.Name(), testLog.SigVerifier, verifiers...) + if err != nil { + t.Fatalf("failed to verify cosigned checkpoint: %v", err) + } + if got, want := len(n.Sigs), 1+tc.numMirrors; got != want { + t.Errorf("note signatures: got %d, want %d", got, want) + } + if got, want := uint64(cp.Size), uint64(size); got != want { + t.Errorf("checkpoint size: got %d, want %d", got, want) + } + }) } } @@ -126,28 +143,37 @@ func mustNewKeypair(t *testing.T, name string) (f_note.Signer, note.Verifier) { return s, v } -func startMockMirror(t *testing.T, signer note.Signer, logVerifier note.Verifier) *httptest.Server { +func startMockMirror(t *testing.T, signer note.Signer, logVerifier note.Verifier, failCount int) *httptest.Server { t.Helper() var mu sync.Mutex var pendingCP []byte + attempts := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + + if strings.HasSuffix(r.URL.Path, "/add-checkpoint") || strings.HasSuffix(r.URL.Path, "/add-entries") { + if attempts < failCount { + attempts++ + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte("mock error")) + return + } + } + if strings.HasSuffix(r.URL.Path, "/add-checkpoint") { body, _ := io.ReadAll(r.Body) parts := bytes.SplitN(body, []byte("\n\n"), 2) if len(parts) == 2 { - mu.Lock() pendingCP = parts[1] - mu.Unlock() } w.WriteHeader(http.StatusOK) return } if strings.HasSuffix(r.URL.Path, "/add-entries") { _, _ = io.Copy(io.Discard, r.Body) - mu.Lock() cp := pendingCP - mu.Unlock() if len(cp) == 0 { w.WriteHeader(http.StatusBadRequest) From 03e53abb15c4bd29fa068cf72a206a3fa6f04b30 Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Fri, 31 Jul 2026 09:15:18 +0000 Subject: [PATCH 4/5] Improvements. --- append_lifecycle.go | 35 +++- internal/mirror/gateway/mirror_gateway.go | 191 +++++++++--------- .../mirror/gateway/mirror_gateway_test.go | 36 ++-- 3 files changed, 145 insertions(+), 117 deletions(-) diff --git a/append_lifecycle.go b/append_lifecycle.go index 8ab1f512c..3fe41b853 100644 --- a/append_lifecycle.go +++ b/append_lifecycle.go @@ -19,7 +19,9 @@ import ( "context" "errors" "fmt" + "maps" "net/http" + "net/url" "os" "slices" "sync" @@ -731,13 +733,40 @@ func (o *AppendOptions) WithAntispam(inMemEntries uint, as Antispam) *AppendOpti return o } +// parseURLs converts a list of URL strings to a list of *url.URL, failing if any cannot be parsed. +func parseURLs(us []string) ([]*url.URL, error) { + ret := make([]*url.URL, 0, len(us)) + for _, s := range us { + u, err := url.Parse(s) + if err != nil { + return nil, err + } + ret = append(ret, u) + } + return ret, nil +} + // CheckpointPublisher returns a function which should be used to create, sign, and potentially witness a new checkpoint. func (o AppendOptions) CheckpointPublisher(ctx context.Context, lr LogReader, httpClient *http.Client) func(context.Context, uint64, []byte) ([]byte, error) { - var gw *m_gateway.Gateway - if len(o.mirrors.Components) > 0 { - gw = m_gateway.NewGateway(ctx, httpClient, o.mirrors, lr, o.primarySigner.Name()) + // TODO(al): Need a better way of surfacing errors. Maybe add a validate() func to AppendOptions? + mirrorURLs, err := parseURLs(slices.Collect(maps.Keys(o.mirrors.WitnessEndpoints()))) + if err != nil { + return func(_ context.Context, _ uint64, _ []byte) ([]byte, error) { + return nil, fmt.Errorf("failed to parse mirror URLs: %w", err) + } + } + gw, err := m_gateway.NewGateway(ctx, m_gateway.Options{ + Mirrors: mirrorURLs, + LogReader: lr, + LogOrigin: o.primarySigner.Name(), + }) + if err != nil { + return func(_ context.Context, _ uint64, _ []byte) ([]byte, error) { + return nil, fmt.Errorf("failed to create gateway: %w", err) + } } + // Now return the actual publisher func. return func(ctx context.Context, size uint64, root []byte) ([]byte, error) { return otel.Trace(ctx, "tessera.CheckpointPublisher", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) { cp, err := o.newCP(ctx, size, root) diff --git a/internal/mirror/gateway/mirror_gateway.go b/internal/mirror/gateway/mirror_gateway.go index ce99f9b16..41d514697 100644 --- a/internal/mirror/gateway/mirror_gateway.go +++ b/internal/mirror/gateway/mirror_gateway.go @@ -17,24 +17,17 @@ package gateway import ( "context" "fmt" - "io" "net/http" "net/url" - "os" - "sync" + "sync/atomic" "time" "log/slog" + "github.com/transparency-dev/tessera/client" "github.com/transparency-dev/tessera/client/mirror" - "golang.org/x/mod/sumdb/note" ) -// WitnessGroup defines the subset of tessera.WitnessGroup methods needed by the gateway. -type WitnessGroup interface { - WitnessEndpoints() map[string][]note.Verifier -} - // LogReader defines the subset of tessera.LogReader methods needed by the gateway. type LogReader interface { ReadTile(ctx context.Context, level, index uint64, p uint8) ([]byte, error) @@ -65,72 +58,59 @@ type Gateway struct { targets []*mirrorTarget } +// Options represents the configuration for a Gateway. +type Options struct { + // HTTPClient is the HTTP client to use for all HTTP operations, if nil uses the DefaultHTTPClient. + HTTPClient *http.Client + // Mirrors defines the pool of mirrors to update. + Mirrors []*url.URL + // LogReader provides access to the main log. + LogReader LogReader + // LogOrigin is the origin ID of the log. + LogOrigin string +} + // NewGateway creates a new Gateway that will keep mirrors up-to-date. -func NewGateway(ctx context.Context, httpClient *http.Client, mirrors WitnessGroup, lr LogReader, logOrigin string) *Gateway { - if httpClient == nil { - httpClient = http.DefaultClient +func NewGateway(ctx context.Context, opts Options) (*Gateway, error) { + if opts.HTTPClient == nil { + slog.WarnContext(ctx, "MirrorGateway:No HTTP client configured, using DefaultHTTPClient") + opts.HTTPClient = http.DefaultClient + } + if opts.LogOrigin == "" { + return nil, fmt.Errorf("log origin is required") + } + if opts.LogReader == nil { + return nil, fmt.Errorf("log reader is required") } g := &Gateway{ - httpClient: httpClient, - lr: lr, + httpClient: opts.HTTPClient, + lr: opts.LogReader, } - endpoints := mirrors.WitnessEndpoints() - for u := range endpoints { - parsedURL, err := url.Parse(u) + endpoints := opts.Mirrors + for _, u := range endpoints { + mirrorFetcher, err := client.NewHTTPFetcher(u, opts.HTTPClient) if err != nil { - slog.ErrorContext(ctx, "Invalid mirror URL", slog.String("url", u), slog.Any("error", err)) - continue - } - - tileFetcher := func(ctx context.Context, level, index uint64, p uint8) ([]byte, error) { - return lr.ReadTile(ctx, level, index, p) - } - bundleFetcher := func(ctx context.Context, index uint64, p uint8) ([]byte, error) { - return lr.ReadEntryBundle(ctx, index, p) - } - mirrorCheckpointFetcher := func(ctx context.Context) ([]byte, error) { - checkpointURL, err := parsedURL.Parse("checkpoint") - if err != nil { - return nil, err - } - req, err := http.NewRequestWithContext(ctx, http.MethodGet, checkpointURL.String(), nil) - if err != nil { - return nil, err - } - resp, err := httpClient.Do(req) - if err != nil { - return nil, err - } - defer func() { - _ = resp.Body.Close() - }() - if resp.StatusCode == http.StatusNotFound { - return nil, os.ErrNotExist - } - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("failed to fetch checkpoint from mirror: status %d", resp.StatusCode) - } - return io.ReadAll(resp.Body) + return nil, fmt.Errorf("invalid mirror URL %v: %v", u, err) } mOpts := mirror.NewOptions(). - WithMirrorURL(parsedURL). - WithHTTPClient(httpClient). - WithLogOrigin(logOrigin). - WithTileFetcher(tileFetcher). - WithBundleFetcher(bundleFetcher). - WithMirrorCheckpointFetcher(mirrorCheckpointFetcher) + WithMirrorURL(u). + WithHTTPClient(opts.HTTPClient). + WithLogOrigin(opts.LogOrigin). + WithTileFetcher(opts.LogReader.ReadTile). + WithBundleFetcher(opts.LogReader.ReadEntryBundle). + WithMirrorCheckpointFetcher(mirrorFetcher.ReadCheckpoint) c, err := mirror.NewClient(ctx, mOpts) if err != nil { - slog.ErrorContext(ctx, "Failed to create mirror client", slog.String("url", u), slog.Any("error", err)) + slog.ErrorContext(ctx, "MirrorGateway: Failed to create mirror client", slog.String("url", u.String()), slog.Any("error", err)) continue } target := &mirrorTarget{ - url: parsedURL, + url: u, client: c, goals: make(chan goal, 1), } @@ -140,19 +120,21 @@ func NewGateway(ctx context.Context, httpClient *http.Client, mirrors WitnessGro go g.runWorker(ctx, target) } - return g + return g, nil } // CosignCheckpoint updates the goals for all mirrors and returns a channel on which it will send // cosignatures as they are successfully fetched from the mirrors. // The channel is closed once all mirrors' signatures have been sent or the context is canceled. func (g *Gateway) CosignCheckpoint(ctx context.Context, cp []byte, cpSize uint64) <-chan []byte { + out := make(chan []byte, len(g.targets)) + if len(g.targets) == 0 { - return nil + close(out) + return out } - out := make(chan []byte, len(g.targets)) - wg := sync.WaitGroup{} + N := &atomic.Uint32{} // Send goals to each of the target workers, but don't block if they're // already busy. @@ -161,12 +143,18 @@ func (g *Gateway) CosignCheckpoint(ctx context.Context, cp []byte, cpSize uint64 cp: cp, cpSize: cpSize, done: func(sig []byte, err error) { - defer wg.Done() + // Last one out please turn off the lights. + defer func() { + if N.Add(1) == uint32(len(g.targets)) { + close(out) + } + }() + if err != nil { - slog.ErrorContext(ctx, "Mirror sync failed", slog.String("url", target.url.String()), slog.Any("error", err)) + slog.ErrorContext(ctx, "MirrorGateway: Sync failed", slog.String("url", target.url.String()), slog.Any("error", err)) return } - slog.InfoContext(ctx, "Mirror sync succeeded", slog.String("url", target.url.String()), slog.Uint64("size", cpSize)) + slog.InfoContext(ctx, "MirrorGateway: Sync succeeded", slog.String("url", target.url.String()), slog.Uint64("size", cpSize)) out <- sig }, } @@ -179,8 +167,6 @@ func (g *Gateway) CosignCheckpoint(ctx context.Context, cp []byte, cpSize uint64 case <-ctx.Done(): done = true case target.goals <- newGoal: - // The goal was sent, we're done - wg.Add(1) done = true default: // No space in the goals channel, try to supercede the goal currently in there. @@ -196,11 +182,6 @@ func (g *Gateway) CosignCheckpoint(ctx context.Context, cp []byte, cpSize uint64 } } - go func() { - wg.Wait() - close(out) - }() - return out } @@ -210,12 +191,15 @@ func (g *Gateway) CosignCheckpoint(ctx context.Context, cp []byte, cpSize uint64 // It will block on the goals channel until a goal is received, or the context is // cancelled. func (g *Gateway) runWorker(ctx context.Context, target *mirrorTarget) { - slog.InfoContext(ctx, "Starting mirror worker", slog.String("url", target.url.String())) - defer slog.InfoContext(ctx, "Stopping mirror worker", slog.String("url", target.url.String())) + slog.InfoContext(ctx, "MirrorGateway: Starting worker", slog.String("url", target.url.String())) + defer slog.InfoContext(ctx, "MirrorGateway: Stopping worker", slog.String("url", target.url.String())) for { select { case <-ctx.Done(): + for job := range target.goals { + job.done(nil, ctx.Err()) + } return case job, ok := <-target.goals: if !ok { @@ -223,32 +207,43 @@ func (g *Gateway) runWorker(ctx context.Context, target *mirrorTarget) { return } - // Loop until the goal is met, or we timeout. - done := false - interval := time.Millisecond - for !done { - select { - case <-ctx.Done(): + goalJob := g.chaseGoal(target, job, 1*time.Minute) + sigs, err := goalJob(ctx) + job.done(sigs, err) + } + } +} + +func (g *Gateway) chaseGoal(target *mirrorTarget, job goal, timeout time.Duration) func(context.Context) ([]byte, error) { + return func(ctx context.Context) ([]byte, error) { + var rSigs []byte + var rErr error + + interval := time.Millisecond + // Loop until the goal is met, or we timeout. + for done := false; !done; { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(interval): + interval = time.Second + } + + // In a func for context defer. + func() { + cctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + slog.DebugContext(cctx, "MirrorGateway: Syncing mirror", slog.String("url", target.url.String()), slog.Uint64("goal", job.cpSize)) + rSigs, rErr = target.client.Sync(cctx, job.cp, job.cpSize) + if rErr != nil { + slog.WarnContext(cctx, "MirrorGateway: Sync failed, retrying", slog.String("url", target.url.String()), slog.Any("error", rErr)) + // not done, loop and retry. return - case <-time.After(interval): - interval = time.Second } - // In a func for context defer. - func() { - // TODO(al): Make this configurable? Should be plenty of time for normal operation, and we'll retry anyway if we do timeout. - cctx, cancel := context.WithTimeout(ctx, 1*time.Minute) - defer cancel() - - slog.DebugContext(cctx, "Syncing mirror", slog.String("url", target.url.String()), slog.Uint64("goal", job.cpSize)) - sigs, err := target.client.Sync(cctx, job.cp, job.cpSize) - if err != nil { - slog.WarnContext(ctx, "Mirror sync attempt failed, retrying", slog.String("url", target.url.String()), slog.Any("error", err)) - return - } - done = true - job.done(sigs, nil) - }() - } + done = true + }() } + return rSigs, rErr } } diff --git a/internal/mirror/gateway/mirror_gateway_test.go b/internal/mirror/gateway/mirror_gateway_test.go index 2de4c017b..6795335cb 100644 --- a/internal/mirror/gateway/mirror_gateway_test.go +++ b/internal/mirror/gateway/mirror_gateway_test.go @@ -21,6 +21,7 @@ import ( "io" "net/http" "net/http/httptest" + "net/url" "slices" "strings" "sync" @@ -35,14 +36,6 @@ import ( "golang.org/x/mod/sumdb/note" ) -type fakeWitnessGroup struct { - endpoints map[string][]note.Verifier -} - -func (f fakeWitnessGroup) WitnessEndpoints() map[string][]note.Verifier { - return f.endpoints -} - func TestGateway(t *testing.T) { for _, tc := range []struct { name string @@ -50,10 +43,12 @@ func TestGateway(t *testing.T) { failCount int }{ { + name: "no mirrors", + numMirrors: 0, + }, { name: "multiple mirrors", numMirrors: 3, - }, - { + }, { name: "retry on transient error", numMirrors: 1, failCount: 2, @@ -86,23 +81,32 @@ func TestGateway(t *testing.T) { } var verifiers []note.Verifier - endpoints := make(map[string][]note.Verifier) + var mirrorURLs []*url.URL for i := range tc.numMirrors { signer, verifier := mustNewKeypair(t, fmt.Sprintf("Mirror-%d", i)) server := startMockMirror(t, signer, testLog.SigVerifier, tc.failCount) defer server.Close() - endpoints[server.URL] = []note.Verifier{verifier} + sURL, err := url.Parse(server.URL) + if err != nil { + t.Fatalf("failed to parse mirror URL: %v", err) + } + mirrorURLs = append(mirrorURLs, sURL) + verifiers = append(verifiers, verifier) } - policy := fakeWitnessGroup{ - endpoints: endpoints, + g, err := gateway.NewGateway(t.Context(), gateway.Options{ + HTTPClient: http.DefaultClient, + Mirrors: mirrorURLs, + LogReader: testLog.LogReader, + LogOrigin: "test", + }) + if err != nil { + t.Fatalf("failed to create gateway: %v", err) } - g := gateway.NewGateway(t.Context(), http.DefaultClient, policy, testLog.LogReader, "test") - // Call CosignCheckpoint and gather signatures. sigCh := g.CosignCheckpoint(t.Context(), goalCP, size) var cosigs []byte From 6fd1a3f702d21c339dda9e58539e575a93872674 Mon Sep 17 00:00:00 2001 From: Al Cutter Date: Fri, 31 Jul 2026 13:53:53 +0000 Subject: [PATCH 5/5] Use mirror in POSIX conformance binary --- cmd/conformance/posix/README.md | 12 +++++++ cmd/conformance/posix/main.go | 39 +++++++++++++++++------ integration/integration_test.go | 4 ++- internal/hammer/hammer.go | 4 +-- internal/mirror/gateway/mirror_gateway.go | 2 +- 5 files changed, 48 insertions(+), 13 deletions(-) diff --git a/cmd/conformance/posix/README.md b/cmd/conformance/posix/README.md index 6dbf465e8..7a3de3e46 100644 --- a/cmd/conformance/posix/README.md +++ b/cmd/conformance/posix/README.md @@ -52,3 +52,15 @@ go run github.com/mhutchinson/woodpecker@main \ --custom_log_origin=example.com/log/testdata \ --custom_log_vkey=${LOG_PUBLIC_KEY} ``` + +## Mirroring + +> [!WARNING] +> Experimental feature, not subject to the SemVer policy! + +This binary supports mirroring the contents of the log to a [tlog-mirror](https://c2sp.org/tlog-mirror) compliant mirror, +such as the one(s) under [cmd/mtc/mirror](/cmd/mtc/mirror). + +Pass the path to a configuration file in the [tlog-policy](https://c2sp.org/tlog-policy) format to the `--mirror_policy` flag +to enable this feature. + diff --git a/cmd/conformance/posix/main.go b/cmd/conformance/posix/main.go index 99f3fe507..14a928d28 100644 --- a/cmd/conformance/posix/main.go +++ b/cmd/conformance/posix/main.go @@ -28,10 +28,11 @@ import ( "path/filepath" "time" - "golang.org/x/mod/sumdb/note" - "log/slog" + f_note "github.com/transparency-dev/formats/note" + "golang.org/x/mod/sumdb/note" + "github.com/transparency-dev/tessera" "github.com/transparency-dev/tessera/storage/posix" badger_as "github.com/transparency-dev/tessera/storage/posix/antispam" @@ -45,6 +46,7 @@ var ( additionalPrivateKeyFiles = []string{} slogLevel = flag.Int("slog_level", 0, "The cut-off threshold for structured logging. Default is 0 (INFO). See https://pkg.go.dev/log/slog#Level for other levels.") logFormat = flag.String("log_format", "text", "The format of the logs: text or json.") + mirrorPolicyFile = flag.String("mirror_policy", "", "File containing the mirror policy.") ) func init() { @@ -93,12 +95,29 @@ func main() { } } - appender, shutdown, _, err := tessera.NewAppender(ctx, driver, tessera.NewAppendOptions(). + opts := tessera.NewAppendOptions(). WithCheckpointSigner(s, a...). WithCheckpointInterval(time.Second). WithCheckpointRepublishInterval(time.Minute). WithBatching(256, time.Second). - WithAntispam(tessera.DefaultAntispamInMemorySize, antispam)) + WithAntispam(tessera.DefaultAntispamInMemorySize, antispam) + if *mirrorPolicyFile != "" { + b, err := os.ReadFile(*mirrorPolicyFile) + if err != nil { + slog.ErrorContext(ctx, "Failed to read mirror policy", slog.Any("error", err)) + os.Exit(1) + } + policy, err := tessera.NewWitnessGroupFromPolicy(b) + if err != nil { + slog.ErrorContext(ctx, "Failed to parse mirror policy", slog.Any("error", err)) + os.Exit(1) + } + opts = opts.WithMirrors(policy, nil) + slog.InfoContext(ctx, "Mirroring enabled", slog.Any("policy", policy)) + } + + appender, shutdown, _, err := tessera.NewAppender(ctx, driver, opts) + if err != nil { slog.ErrorContext(ctx, "Failed to create new appender", slog.Any("error", err)) os.Exit(1) @@ -163,7 +182,7 @@ func getSignersOrDie() (note.Signer, []note.Signer) { slog.ErrorContext(context.Background(), "Unable to get additional private key", slog.String("file", p), slog.Any("error", err)) os.Exit(1) } - k, err := note.NewSigner(kr) + k, err := f_note.NewSignerForCosignatureV1(kr) if err != nil { slog.ErrorContext(context.Background(), "Failed to instantiate signer", slog.String("file", p), slog.Any("error", err)) os.Exit(1) @@ -190,10 +209,12 @@ func getSignerOrDie() note.Signer { os.Exit(1) } } - s, err := note.NewSigner(privKey) - if err != nil { - slog.ErrorContext(context.Background(), "Failed to instantiate signer", slog.Any("error", err)) - os.Exit(1) + var s note.Signer + if s, err = note.NewSigner(privKey); err != nil { + if s, err = f_note.NewSignerForCosignatureV1(privKey); err != nil { + slog.ErrorContext(context.Background(), "Failed to instantiate signer", slog.Any("error", err)) + os.Exit(1) + } } return s } diff --git a/integration/integration_test.go b/integration/integration_test.go index 755f7f8c7..4c9291cf1 100644 --- a/integration/integration_test.go +++ b/integration/integration_test.go @@ -34,10 +34,12 @@ import ( "log/slog" + f_note "github.com/transparency-dev/formats/note" "github.com/transparency-dev/merkle/proof" "github.com/transparency-dev/merkle/rfc6962" "github.com/transparency-dev/tessera/api/layout" "github.com/transparency-dev/tessera/client" + "golang.org/x/mod/sumdb/note" "golang.org/x/sync/errgroup" ) @@ -74,7 +76,7 @@ func TestMain(m *testing.M) { } var err error - noteVerifier, err = note.NewVerifier(*logPublicKey) + noteVerifier, err = f_note.NewVerifier(*logPublicKey) if err != nil { slog.ErrorContext(context.Background(), "Failed to create new verifier", slog.Any("error", err)) os.Exit(1) diff --git a/internal/hammer/hammer.go b/internal/hammer/hammer.go index e6bf07acc..6a97d0c07 100644 --- a/internal/hammer/hammer.go +++ b/internal/hammer/hammer.go @@ -35,7 +35,7 @@ import ( "github.com/transparency-dev/tessera/client" "github.com/transparency-dev/tessera/internal/hammer/loadtest" - "golang.org/x/mod/sumdb/note" + f_note "github.com/transparency-dev/formats/note" "golang.org/x/net/http2" "log/slog" @@ -108,7 +108,7 @@ func main() { ctx, cancel := context.WithCancel(context.Background()) - logSigV, err := note.NewVerifier(*logPubKey) + logSigV, err := f_note.NewVerifier(*logPubKey) if err != nil { slog.ErrorContext(ctx, "failed to create verifier", slog.Any("error", err)) os.Exit(1) diff --git a/internal/mirror/gateway/mirror_gateway.go b/internal/mirror/gateway/mirror_gateway.go index 41d514697..ef2549ce1 100644 --- a/internal/mirror/gateway/mirror_gateway.go +++ b/internal/mirror/gateway/mirror_gateway.go @@ -220,7 +220,7 @@ func (g *Gateway) chaseGoal(target *mirrorTarget, job goal, timeout time.Duratio var rErr error interval := time.Millisecond - // Loop until the goal is met, or we timeout. + // Loop until the goal is met, or the context is done. for done := false; !done; { select { case <-ctx.Done():