Implement mirror support in append lifecycle

append_lifecycle.go

@@ -15,10 +15,13 @@
 package tessera
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"maps"
 	"net/http"
+	"net/url"
 	"os"
 	"slices"
 	"sync"
@@ -30,6 +33,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"
@@ -729,8 +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(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) {
+	// 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)
@@ -748,8 +784,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 +853,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
 	})
 }
 

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
+		}
+	}
+}

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.