diff --git a/append_lifecycle.go b/append_lifecycle.go index f78474edb..d13094143 100644 --- a/append_lifecycle.go +++ b/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 { diff --git a/internal/witness/client.go b/internal/witness/client.go index 083792b6f..f5dc61f13 100644 --- a/internal/witness/client.go +++ b/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 } } diff --git a/internal/witness/client_test.go b/internal/witness/client_test.go index fea0116bb..acdd791f2 100644 --- a/internal/witness/client_test.go +++ b/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,7 +368,7 @@ 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) } @@ -374,9 +376,8 @@ func TestWitness_UpdateResponse(t *testing.T) { 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) } }) }