Make witness client return sigs only

append_lifecycle.go

@@ -735,51 +735,65 @@ func (o AppendOptions) CheckpointPublisher(lr LogReader, httpClient *http.Client
 			span.AddEvent("Created CP")
 			appenderSignedSize.Record(ctx, otel.Clamp64(size))
 
-			span.AddEvent("Starting witnessing")
-			// Handle witnessing
-			if len(o.witnesses.Components) != 0 {
-				// Figure out the likely size the witnesses are aware of, but don't fail hard if we're unable
-				// to do so:
-				// a) it could be that this is the first checkpoint we're publishing
-				// b) the witnessing protocol has a fallback path in case we get it wrong, anyway.
-				var oldSize uint64
-				oldCP, err := lr.ReadCheckpoint(ctx)
-				if err != nil {
-					slog.InfoContext(ctx, "Failed to fetch old checkpoint", slog.Any("error", err))
-				} else {
-					_, oldSize, _, err = parse.CheckpointUnsafe(oldCP)
-					if err != nil {
-						return nil, fmt.Errorf("failed to parse old checkpoint: %v", err)
-					}
-				}
-				wg := witness.NewWitnessGateway(o.witnesses, httpClient, oldSize, lr.ReadTile)
+			wSigs, err := witnessCheckpoint(ctx, cp, size, o.witnesses, lr, httpClient, o.witnessOpts)
+			if err != nil {
+				return nil, err
+			}
 
-				start := time.Now()
+			cp = append(cp, wSigs...)
 
-				ctx, cancel := context.WithTimeout(ctx, o.witnessOpts.Timeout)
-				defer cancel()
+			return cp, nil
+		})
+	}
+}
 
-				span.AddEvent("Sending witness requests")
-				witAttr := []attribute.KeyValue{}
-				cp, err = wg.Witness(ctx, cp)
-				if err != nil {
-					if !o.witnessOpts.FailOpen {
-						appenderWitnessRequests.Add(ctx, 1, metric.WithAttributes(attribute.String("error.type", "failed")))
-						return nil, err
-					}
-					slog.WarnContext(ctx, "WitnessGateway: failing-open despite error", slog.Any("error", err))
-					witAttr = append(witAttr, attribute.String("error.type", "failed_open"))
-				}
+// witnessCheckpoint takes care of witnessing the given checkpoint with the provided witness policy.
+// Returns signatures from witnesses, ready to append to the checkpoint, or an error.
+func witnessCheckpoint(ctx context.Context, cp []byte, cpSize uint64, witnesses WitnessGroup, lr LogReader, httpClient *http.Client, opts WitnessOptions) ([]byte, error) {
+	return otel.Trace(ctx, "tessera.witnessCheckpoint", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
+		if len(witnesses.Components) == 0 {
+			return nil, nil
+		}
+		span.AddEvent("Starting witnessing")
+		// Figure out the likely size the witnesses are aware of, but don't fail hard if we're unable
+		// to do so:
+		// a) it could be that this is the first checkpoint we're publishing
+		// b) the witnessing protocol has a fallback path in case we get it wrong, anyway.
+		var oldSize uint64
+		oldCP, err := lr.ReadCheckpoint(ctx)
+		if err != nil {
+			slog.InfoContext(ctx, "Failed to fetch old checkpoint", slog.Any("error", err))
+		} else {
+			_, oldSize, _, err = parse.CheckpointUnsafe(oldCP)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse old checkpoint: %v", err)
+			}
+		}
+		wg := witness.NewWitnessGateway(witnesses, httpClient, oldSize, lr.ReadTile)
+
+		start := time.Now()
+
+		ctx, cancel := context.WithTimeout(ctx, opts.Timeout)
+		defer cancel()
 
-				appenderWitnessRequests.Add(ctx, 1, metric.WithAttributes(witAttr...))
-				appenderWitnessedSize.Record(ctx, otel.Clamp64(size))
-				d := time.Since(start)
-				appenderWitnessHistogram.Record(ctx, d.Milliseconds(), metric.WithAttributes(witAttr...))
+		span.AddEvent("Sending witness requests")
+		witAttr := []attribute.KeyValue{}
+		cpSigs, err := wg.Witness(ctx, cp)
+		if err != nil {
+			if !opts.FailOpen {
+				appenderWitnessRequests.Add(ctx, 1, metric.WithAttributes(attribute.String("error.type", "failed")))
+				return nil, err
 			}
+			slog.WarnContext(ctx, "WitnessGateway: failing-open despite error", slog.Any("error", err))
+			witAttr = append(witAttr, attribute.String("error.type", "failed_open"))
+		}
 
-			return cp, nil
-		})
-	}
+		appenderWitnessRequests.Add(ctx, 1, metric.WithAttributes(witAttr...))
+		appenderWitnessedSize.Record(ctx, otel.Clamp64(cpSize))
+		d := time.Since(start)
+		appenderWitnessHistogram.Record(ctx, d.Milliseconds(), metric.WithAttributes(witAttr...))
+		return cpSigs, nil
+	})
 }
 
 func (o AppendOptions) BatchMaxAge() time.Duration {

internal/witness/client.go

@@ -169,12 +169,12 @@ type WitnessGateway struct {
 }
 
 // Witness sends out a new checkpoint (which must be signed by the log), to all witnesses
-// and returns the checkpoint as soon as the policy the WitnessGateway was constructed with
+// and returns gathered cosignatures as soon as the policy the WitnessGateway was constructed with
 // is Satisfied.
 func (wg *WitnessGateway) Witness(ctx context.Context, cp []byte) ([]byte, error) {
 	return otel.Trace(ctx, "tessera.witnessgateway.Witness", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
 		if len(wg.witnesses) == 0 {
-			return cp, nil
+			return nil, nil
 		}
 		ctx, cancel := context.WithCancel(ctx)
 		defer cancel()
@@ -227,7 +227,6 @@ func (wg *WitnessGateway) Witness(ctx context.Context, cp []byte) ([]byte, error
 		span.AddEvent("Waiting for signatures")
 		// Consume the results coming back from each witness
 		var sigBlock bytes.Buffer
-		sigBlock.Write(cp)
 		for r := range results {
 			if r.err != nil {
 				err = errors.Join(err, r.err)
@@ -245,9 +244,9 @@ func (wg *WitnessGateway) Witness(ctx context.Context, cp []byte) ([]byte, error
 			sigBlock.Write(r.sig)
 
 			// See whether the group is satisfied now
-			if newCp := sigBlock.Bytes(); wg.group.Satisfied(newCp) {
+			if newCp := append(slices.Clone(cp), sigBlock.Bytes()...); wg.group.Satisfied(newCp) {
 				span.AddEvent("Policy satisfied")
-				return newCp, nil
+				return sigBlock.Bytes(), nil
 			}
 		}
 

internal/witness/client_test.go

@@ -24,6 +24,7 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"slices"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -186,13 +187,14 @@ func TestWitnessGateway_Update(t *testing.T) {
 
 			g := witness.NewWitnessGateway(tC.group, ts.Client(), 0, testLogTileFetcher)
 
-			witnessedCP, err := g.Witness(ctx, logSignedCheckpoint)
+			cpSigs, err := g.Witness(ctx, logSignedCheckpoint)
 			if got, want := err != nil, tC.wantErr; got != want {
 				t.Fatalf("got != want (%t != %t): %v", got, want, err)
 			}
 			if tC.wantErr {
 				return
 			}
+			witnessedCP := append(slices.Clone(logSignedCheckpoint), cpSigs...)
 			n, err := note.Open(witnessedCP, note.VerifierList(logVerifier, wit1.Key, wit2.Key))
 			if err != nil {
 				t.Fatalf("failed to open note %q: %v", witnessedCP, err)
@@ -366,17 +368,16 @@ func TestWitness_UpdateResponse(t *testing.T) {
 				t.Fatal(err)
 			}
 			g := witness.NewWitnessGateway(tessera.NewWitnessGroup(1, wit1), ts.Client(), 0, testLogTileFetcher)
-			witnessed, err := g.Witness(ctx, logSignedCheckpoint)
+			cpSigs, err := g.Witness(ctx, logSignedCheckpoint)
 			if got, want := err != nil, tC.wantErr; got != want {
 				t.Fatalf("got != want (%t != %t): %v", got, want, err)
 			}
 			if tC.wantErr {
 				return
 			}
 
-			sigs := witnessed[len(logSignedCheckpoint):]
-			if !bytes.Equal(sigs, tC.wantResult) {
-				t.Errorf("expected result %q but got %q", tC.body, sigs)
+			if !bytes.Equal(cpSigs, tC.wantResult) {
+				t.Errorf("expected result %q but got %q", tC.body, cpSigs)
 			}
 		})
 	}