append_lifecycle.go

// Copyright 2024 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 tessera

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"slices"
	"sync"
	"sync/atomic"
	"time"

	"log/slog"

	f_log "github.com/transparency-dev/formats/log"
	"github.com/transparency-dev/formats/policy"
	"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"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
	"go.opentelemetry.io/otel/trace"
	"golang.org/x/mod/sumdb/note"
	"golang.org/x/sync/errgroup"
)

const (
	// DefaultBatchMaxSize is used by storage implementations if no WithBatching option is provided when instantiating it.
	DefaultBatchMaxSize = 256
	// DefaultBatchMaxAge is used by storage implementations if no WithBatching option is provided when instantiating it.
	DefaultBatchMaxAge = 250 * time.Millisecond
	// DefaultCheckpointInterval is used by storage implementations if no WithCheckpointInterval option is provided when instantiating it.
	DefaultCheckpointInterval = 10 * time.Second
	// DefaultCheckpointRepublishInterval is used by storage implementations if no WithCheckpointRepublishInterval option is provided when instantiating it.
	DefaultCheckpointRepublishInterval = 10 * time.Minute
	// DefaultCheckpointPublicationTimeout is used by storage implementations if no WithCheckpointPublicationTimeout option is provided when instantiating it.
	DefaultCheckpointPublicationTimeout = 5 * time.Second
	// DefaultPushbackMaxOutstanding is used by storage implementations if no WithPushback option is provided when instantiating it.
	DefaultPushbackMaxOutstanding = 4096
	// DefaultGarbageCollectionInterval is the default value used if no WithGarbageCollectionInterval option is provided.
	DefaultGarbageCollectionInterval = time.Minute
	// DefaultAntispamInMemorySize is the recommended default limit on the number of entries in the in-memory antispam cache.
	// The amount of data stored for each entry is small (32 bytes of hash + 8 bytes of index), so in the general case it should be fine
	// to have a very large cache.
	DefaultAntispamInMemorySize = 256 << 10
	// DefaultWitnessTimeout is the default maximum time to wait for responses from configured witnesses.
	DefaultWitnessTimeout = 1 * time.Second
	// DefaultMirrorTimeout is the default maximum time to wait for responses from configured mirrors.
	DefaultMirrorTimeout = 10 * time.Second
	// DefaultEntrySizeLimit is the maximum possible size of data for a single entry, as specified by C2SP tlog-tiles.
	DefaultEntrySizeLimit = 1<<16 - 1

	// DefaultShutdownTimeout is the maximum duration to wait for the shutdown process to complete.
	DefaultShutdownTimeout = 60 * time.Second
)

var (
	appenderAddsTotal         metric.Int64Counter
	appenderAddHistogram      metric.Int64Histogram
	appenderHighestIndex      metric.Int64Gauge
	appenderIntegratedSize    metric.Int64Gauge
	appenderIntegrateLatency  metric.Int64Histogram
	appenderDeadlineRemaining metric.Int64Histogram
	appenderNextIndex         metric.Int64Gauge
	appenderSignedSize        metric.Int64Gauge
	appenderWitnessedSize     metric.Int64Gauge
	appenderWitnessRequests   metric.Int64Counter
	appenderWitnessHistogram  metric.Int64Histogram

	followerEntriesProcessed metric.Int64Gauge
	followerLag              metric.Int64Gauge

	// Custom histogram buckets as we're still interested in details in the 1-2s area.
	histogramBuckets = []float64{0, 10, 50, 100, 200, 300, 400, 500, 600, 700, 800, 900, 1000, 1200, 1400, 1600, 1800, 2000, 2500, 3000, 4000, 5000, 6000, 8000, 10000}
)

func init() {
	ctx := context.Background()
	var err error

	appenderAddsTotal, err = meter.Int64Counter(
		"tessera.appender.add.calls",
		metric.WithDescription("Number of calls to the appender lifecycle Add function"),
		metric.WithUnit("{call}"))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderAddsTotal metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderAddHistogram, err = meter.Int64Histogram(
		"tessera.appender.add.duration",
		metric.WithDescription("Duration of calls to the appender lifecycle Add function"),
		metric.WithUnit("ms"),
		metric.WithExplicitBucketBoundaries(histogramBuckets...))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderAddDuration metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderHighestIndex, err = meter.Int64Gauge(
		"tessera.appender.index",
		metric.WithDescription("Highest index assigned by appender lifecycle Add function"))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderHighestIndex metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderIntegratedSize, err = meter.Int64Gauge(
		"tessera.appender.integrated.size",
		metric.WithDescription("Size of the integrated (but not necessarily published) tree"),
		metric.WithUnit("{entry}"))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderIntegratedSize metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderIntegrateLatency, err = meter.Int64Histogram(
		"tessera.appender.integrate.latency",
		metric.WithDescription("Duration between an index being assigned by Add, and that index being integrated in the tree"),
		metric.WithUnit("ms"),
		metric.WithExplicitBucketBoundaries(histogramBuckets...))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderIntegrateLatency metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderDeadlineRemaining, err = meter.Int64Histogram(
		"tessera.appender.deadline.remaining",
		metric.WithDescription("Duration remaining before context cancellation when appender is invoked (only set for contexts with deadline)"),
		metric.WithUnit("ms"),
		metric.WithExplicitBucketBoundaries(histogramBuckets...))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderDeadlineRemaining metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderNextIndex, err = meter.Int64Gauge(
		"tessera.appender.next_index",
		metric.WithDescription("The next available index to be assigned to entries"))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderNextIndex metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderSignedSize, err = meter.Int64Gauge(
		"tessera.appender.signed.size",
		metric.WithDescription("Size of the latest signed checkpoint"),
		metric.WithUnit("{entry}"))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderSignedSize metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderWitnessedSize, err = meter.Int64Gauge(
		"tessera.appender.witnessed.size",
		metric.WithDescription("Size of the latest successfully witnessed checkpoint"),
		metric.WithUnit("{entry}"))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderWitnessedSize metric", slog.Any("error", err))
		os.Exit(1)
	}

	followerEntriesProcessed, err = meter.Int64Gauge(
		"tessera.follower.processed",
		metric.WithDescription("Number of entries processed"),
		metric.WithUnit("{entry}"))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create followerEntriesProcessed metric", slog.Any("error", err))
		os.Exit(1)
	}

	followerLag, err = meter.Int64Gauge(
		"tessera.follower.lag",
		metric.WithDescription("Number of unprocessed entries in the current integrated tree"),
		metric.WithUnit("{entry}"))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create followerLag metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderWitnessRequests, err = meter.Int64Counter(
		"tessera.appender.witness.requests",
		metric.WithDescription("Number of attempts to witness a log checkpoint"),
		metric.WithUnit("{call}"))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderWitnessRequests metric", slog.Any("error", err))
		os.Exit(1)
	}

	appenderWitnessHistogram, err = meter.Int64Histogram(
		"tessera.appender.witness.duration",
		metric.WithDescription("Duration of calls to the configured witness group"),
		metric.WithUnit("ms"),
		metric.WithExplicitBucketBoundaries(histogramBuckets...))
	if err != nil {
		slog.ErrorContext(ctx, "Failed to create appenderWitnessHistogram metric", slog.Any("error", err))
		os.Exit(1)
	}

}

// AddFn adds a new entry to be sequenced by the storage implementation.
//
// This method should quickly return an IndexFuture, which can be called to resolve to the
// index **durably** assigned to the new entry (or an error).
//
// Implementations MUST NOT allow the future to resolve to an index value unless/until it has
// been durably committed by the storage.
//
// Callers MUST NOT assume that an entry has been accepted or durably stored until they have
// successfully resolved the future.
//
// Once the future resolves and returns an index, the entry can be considered to have been
// durably sequenced and will be preserved even in the event that the process terminates.
//
// Once an entry is sequenced, the storage implementation MUST integrate it into the tree soon
// (how long this is expected to take is left unspecified, but as a guideline it should happen
// within single digit seconds). Until the entry is integrated and published, clients of the log
// will not be able to verifiably access this value.
//
// Personalities which require blocking until the entry is integrated (e.g. because they wish
// to return an inclusion proof) may use the PublicationAwaiter to wrap the call to this method.
type AddFn func(ctx context.Context, entry *Entry) IndexFuture

// IndexFuture is the signature of a function which can return an assigned index or error.
//
// Implementations of this func are likely to be "futures", or a promise to return this data at
// some point in the future, and as such will block when called if the data isn't yet available.
type IndexFuture func() (Index, error)

// Index represents a durably assigned index for some entry.
type Index struct {
	// Index is the location in the log to which a particular entry has been assigned.
	Index uint64
	// IsDup is true if Index represents a previously assigned index for an identical entry.
	IsDup bool
}

// Appender allows personalities access to the lifecycle methods associated with logs
// in sequencing mode. This only has a single method, but other methods are likely to be added
// such as a Shutdown method for #341.
type Appender struct {
	Add AddFn
}

// NewAppender returns an Appender, which allows a personality to incrementally append new
// leaves to the log and to read from it.
//
// The return values are the Appender for adding new entries, a shutdown function, a log reader,
// and an error if any of the objects couldn't be constructed.
//
// Shutdown ensures that all calls to Add that have returned a value will be resolved. Any
// futures returned by _this appender_ which resolve to an index will be integrated and have
// a checkpoint that commits to them published if this returns successfully. After this returns,
// any calls to Add will fail.
//
// The context passed into this function will be referenced by any background tasks that are started
// in the Appender. The correct process for shutting down an Appender cleanly is to first call the
// shutdown function that is returned, and then cancel the context. Cancelling the context without calling
// shutdown first may mean that some entries added by this appender aren't in the log when the process
// exits.
func NewAppender(ctx context.Context, d Driver, opts *AppendOptions) (*Appender, func(ctx context.Context) error, LogReader, error) {
	type appendLifecycle interface {
		Appender(context.Context, *AppendOptions) (*Appender, LogReader, error)
	}
	lc, ok := d.(appendLifecycle)
	if !ok {
		return nil, nil, nil, fmt.Errorf("driver %T does not implement Appender lifecycle", d)
	}
	if opts == nil {
		return nil, nil, nil, errors.New("opts cannot be nil")
	}
	slog.InfoContext(ctx, "Creating new appender", slog.Any("options", opts))
	if err := opts.valid(ctx); err != nil {
		return nil, nil, nil, err
	}
	a, r, err := lc.Appender(ctx, opts)
	if err != nil {
		return nil, nil, nil, fmt.Errorf("failed to init appender lifecycle: %v", err)
	}
	for i := len(opts.addDecorators) - 1; i >= 0; i-- {
		a.Add = opts.addDecorators[i](a.Add)
	}
	a.Add = entrySizeLimitDecorator(a.Add, opts.maxEntrySize)
	sd := &integrationStats{}
	a.Add = sd.statsDecorator(a.Add)
	for _, f := range opts.followers {
		go f.Follow(ctx, r)
		go followerStats(ctx, f, r.IntegratedSize)
	}
	go sd.updateStats(ctx, r)
	t := terminator{
		delegate:        a.Add,
		readCheckpoint:  r.ReadCheckpoint,
		followers:       opts.followers,
		shutdownTimeout: opts.shutdownTimeout,
	}
	// TODO(mhutchinson): move this into the decorators
	a.Add = func(ctx context.Context, entry *Entry) IndexFuture {
		if deadline, ok := ctx.Deadline(); ok {
			appenderDeadlineRemaining.Record(ctx, time.Until(deadline).Milliseconds())
		}
		ctx, span := tracer.Start(ctx, "tessera.Appender.Add")
		defer span.End()

		// NOTE: We memoize the returned value here so that repeated calls to the returned
		//		 future don't result in unexpected side-effects from inner AddFn functions
		//		 being called multiple times.
		//		 Currently this is the outermost wrapping of Add so we do the memoization
		//		 here, if this changes, ensure that we move the memoization call so that
		//		 this remains true.
		return memoizeFuture(t.Add(ctx, entry))
	}
	return a, t.Shutdown, r, nil
}

// entrySizeLimitDecorator wraps a delegate AddFn with logic which will return an error if
// it is called with an entry larger than the provided max size.
func entrySizeLimitDecorator(d AddFn, maxSize uint) AddFn {
	return func(ctx context.Context, entry *Entry) IndexFuture {
		if sz := uint(len(entry.Data())); sz > maxSize {
			return func() (Index, error) {
				return Index{}, fmt.Errorf("entry data too large (%d > %d)", sz, maxSize)
			}
		}
		return d(ctx, entry)
	}
}

// memoizeFuture wraps an AddFn delegate with logic to ensure that the delegate is called at most
// once.
func memoizeFuture(delegate IndexFuture) IndexFuture {
	f := sync.OnceValues(func() (Index, error) {
		return delegate()
	})
	return f
}

func followerStats(ctx context.Context, f Follower, size func(context.Context) (uint64, error)) {
	name := f.Name()
	t := time.NewTicker(200 * time.Millisecond)
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}

		n, err := f.EntriesProcessed(ctx)
		if err != nil {
			slog.ErrorContext(ctx, "followerStats: error in EntriesProcessed", slog.String("name", name), slog.Any("error", err))
			continue
		}
		s, err := size(ctx)
		if err != nil {
			slog.ErrorContext(ctx, "followerStats: error in follower size", slog.String("name", name), slog.Any("error", err))
		}
		attrs := metric.WithAttributes(followerNameKey.String(name))
		followerEntriesProcessed.Record(ctx, otel.Clamp64(n), attrs)
		followerLag.Record(ctx, otel.Clamp64(s-n), attrs)
	}
}

// idxAt represents an index first seen at a particular time.
type idxAt struct {
	idx uint64
	at  time.Time
}

// integrationStats knows how to track and populate metrics related to integration performance.
//
// Currently, this tracks integration latency only.
// The integration latency tracking works via a "sample & consume" mechanism, whereby an Add decorator
// will record an assigned index along with the time it was assigned. An asynchronous process will
// periodically compare the sample with the current integrated tree size, and if the sampled index is
// found to be covered by the tree the elapsed period is recorded and the sample "consumed".
//
// Only one sample may be held at a time.
type integrationStats struct {
	// indexSample points to a sampled indexAt, or nil if there has been no sample made _or_ the sample was consumed.
	indexSample atomic.Pointer[idxAt]
}

// sample creates a new sample with the provided index if no sample is already held.
func (i *integrationStats) sample(idx uint64) {
	i.indexSample.CompareAndSwap(nil, &idxAt{idx: idx, at: time.Now()})
}

// latency will check whether the provided tree size is larger than the currently sampled index (if one exists),
// and, if so, "consume" the sample and return the elapsed interval since the sample was taken.
//
// The returned bool is true if a sample exists and whose index is lower than the provided tree size, and
// false otherwise.
func (i *integrationStats) latency(size uint64) (time.Duration, bool) {
	ia := i.indexSample.Load()
	// If there _is_ a sample...
	if ia != nil {
		// and the sampled index is lower than the tree size
		if ia.idx < size {
			// then reset the sample store here so that we're able to accept a future sample.
			i.indexSample.Store(nil)
		}
		return time.Since(ia.at), true
	}
	return 0, false
}

// updateStates periodically checks the current integrated tree size and attempts to
// consume any held sample, updating the metric if possible.
//
// This is a long running function, exitingly only when the provided context is done.
func (i *integrationStats) updateStats(ctx context.Context, r LogReader) {
	if r == nil {
		slog.WarnContext(ctx, "updateStates: nil logreader provided, not updating stats")
		return
	}
	t := time.NewTicker(100 * time.Millisecond)
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
		s, err := r.IntegratedSize(ctx)
		if err != nil {
			slog.ErrorContext(ctx, "Error calling IntegratedSize", slog.Any("error", err))
			continue
		}
		appenderIntegratedSize.Record(ctx, otel.Clamp64(s))
		if d, ok := i.latency(s); ok {
			appenderIntegrateLatency.Record(ctx, d.Milliseconds())
		}
		i, err := r.NextIndex(ctx)
		if err != nil {
			slog.ErrorContext(ctx, "Error calling NextIndex", slog.Any("error", err))
		}
		appenderNextIndex.Record(ctx, otel.Clamp64(i))
	}
}

// statsDecorator wraps a delegate AddFn with code to calculate/update
// metric stats.
func (i *integrationStats) statsDecorator(delegate AddFn) AddFn {
	return func(ctx context.Context, entry *Entry) IndexFuture {
		start := time.Now()
		f := delegate(ctx, entry)

		return func() (Index, error) {
			idx, err := f()
			attr := []attribute.KeyValue{}

			if err != nil {
				switch {
				// Record the fact there was pushback, if any.
				case errors.Is(err, ErrPushbackAntispam):
					attr = append(attr, attribute.String("tessera.pushback", "antispam"))
				case errors.Is(err, ErrPushbackIntegration):
					attr = append(attr, attribute.String("tessera.pushback", "integration"))
				case errors.Is(err, ErrPushback):
					attr = append(attr, attribute.String("tessera.pushback", "other"))
				default:
					// If it's not a pushback, just flag that it's an errored request to avoid high cardinality of attribute values.
					// TODO(al): We might want to bucket errors into OTel status codes in the future, though.
					attr = append(attr, attribute.String("tessera.error.type", "_OTHER"))
				}
			}

			attr = append(attr, attribute.Bool("tessera.duplicate", idx.IsDup))

			appenderAddsTotal.Add(ctx, 1, metric.WithAttributes(attr...))
			d := time.Since(start)
			appenderAddHistogram.Record(ctx, d.Milliseconds(), metric.WithAttributes(attr...))

			if !idx.IsDup {
				i.sample(idx.Index)
			}

			return idx, err
		}
	}
}

type terminator struct {
	delegate       AddFn
	readCheckpoint func(ctx context.Context) ([]byte, error)
	followers      []Follower

	// This mutex guards the stopped state. We use this instead of an atomic.Boolean
	// to get the property that no readers of this state can have the lock when the
	// write gets it. This means that no in-flight Add operations will be occurring on
	// Shutdown.
	mu      sync.RWMutex
	stopped bool

	// wantTreeSize tracks a tree size which is large enough to cover all
	// locally added entries.
	wantTreeSize atomic.Uint64

	// shutdownTimeout is the maximum duration to wait for the shutdown process to complete.
	shutdownTimeout time.Duration
}

func (t *terminator) Add(ctx context.Context, entry *Entry) IndexFuture {
	t.mu.RLock()
	defer t.mu.RUnlock()
	if t.stopped {
		return func() (Index, error) {
			return Index{}, errors.New("appender has been shut down")
		}
	}
	res := t.delegate(ctx, entry)
	return func() (Index, error) {
		i, err := res()
		if err != nil {
			return i, err
		}

		// https://github.com/golang/go/issues/63999 - atomically set minimum tree size for shutdown.
		wantSize := i.Index + 1
		old := t.wantTreeSize.Load()
		for old < wantSize && !t.wantTreeSize.CompareAndSwap(old, wantSize) {
			old = t.wantTreeSize.Load()
		}
		appenderHighestIndex.Record(ctx, otel.Clamp64(t.wantTreeSize.Load()-1))

		return i, err
	}
}

// Shutdown ensures that all calls to Add that have returned a value will be resolved.
// Futures returned by _this appender_ which resolve to an index will be integrated and have
// a checkpoint that commits to them published if this returns successfully.
// Once published, it ensures that all followers are caught up to the largest index.
//
// After this returns, any calls to Add will fail.
func (t *terminator) Shutdown(ctx context.Context) error {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.stopped = true

	wantSize := t.wantTreeSize.Load()
	if wantSize == 0 {
		// special case no work done
		return nil
	}

	timeoutCtx, cancel := context.WithTimeout(ctx, t.shutdownTimeout)
	defer cancel()

	g, gCtx := errgroup.WithContext(timeoutCtx)

	// Wait for checkpoint.
	g.Go(func() error {
		ticker := time.NewTicker(1 * time.Nanosecond)
		defer ticker.Stop()
		for {
			select {
			case <-gCtx.Done():
				return gCtx.Err()
			case <-ticker.C:
				ticker.Reset(100 * time.Millisecond)
			}

			cp, err := t.readCheckpoint(gCtx)
			if err != nil {
				if !errors.Is(err, os.ErrNotExist) {
					return err
				}
				continue
			}
			_, size, _, err := parse.CheckpointUnsafe(cp)
			if err != nil {
				return err
			}
			slog.DebugContext(gCtx, "Shutting down, waiting for checkpoint", slog.Uint64("goal", wantSize), slog.Uint64("current", size))
			if size >= wantSize {
				return nil
			}
		}
	})

	// Wait for followers to catch up to the largest index.
	for _, f := range t.followers {
		g.Go(func() error {
			ticker := time.NewTicker(1 * time.Nanosecond)
			defer ticker.Stop()
			for {
				select {
				case <-gCtx.Done():
					return gCtx.Err()
				case <-ticker.C:
					ticker.Reset(100 * time.Millisecond)
				}

				processed, err := f.EntriesProcessed(gCtx)
				if err != nil {
					return err
				}
				slog.DebugContext(gCtx, "Shutting down, waiting for follower", slog.String("follower", f.Name()), slog.Uint64("goal", wantSize), slog.Uint64("processed", processed))
				if processed >= wantSize {
					return nil
				}
			}
		})
	}

	return g.Wait()
}

// NewAppendOptions creates a new options struct for configuring appender lifecycle instances.
//
// These options are configured through the use of the various `With.*` function calls on the returned
// instance.
func NewAppendOptions() *AppendOptions {
	return &AppendOptions{
		batchMaxSize:                 DefaultBatchMaxSize,
		batchMaxAge:                  DefaultBatchMaxAge,
		entriesPath:                  layout.EntriesPath,
		maxEntrySize:                 DefaultEntrySizeLimit,
		bundleIDHasher:               defaultIDHasher,
		checkpointInterval:           DefaultCheckpointInterval,
		checkpointRepublishInterval:  DefaultCheckpointRepublishInterval,
		checkpointPublicationTimeout: DefaultCheckpointPublicationTimeout,
		addDecorators:                make([]func(AddFn) AddFn, 0),
		pushbackMaxOutstanding:       DefaultPushbackMaxOutstanding,
		garbageCollectionInterval:    DefaultGarbageCollectionInterval,
		shutdownTimeout:              DefaultShutdownTimeout,
	}
}

// AppendOptions holds settings for all storage implementations.
type AppendOptions struct {
	// newCP knows how to format and sign checkpoints.
	newCP func(ctx context.Context, size uint64, hash []byte) ([]byte, error)

	batchMaxAge  time.Duration
	batchMaxSize uint

	pushbackMaxOutstanding uint

	// EntriesPath knows how to format entry bundle paths.
	entriesPath func(n uint64, p uint8) string

	// maxEntrySize is the maximum permitted size of individual entries to be added.
	// By default this will correspond with the limit set out in the C2SP tlog-tiles spec, however
	// it will be overridden when Tessera is being used to implement static-ct.
	//
	// Note that storage implementations MAY enforce their own, lower, limits on entry size due to
	// storage-level constraints. This should be handled internally by those storage implementations,
	// e.g. in their Add() function implementations.
	maxEntrySize uint

	// bundleIDHasher knows how to create antispam leaf identities for entries in a serialised bundle.
	bundleIDHasher func([]byte) ([][]byte, error)

	checkpointInterval           time.Duration
	checkpointRepublishInterval  time.Duration
	checkpointPublicationTimeout time.Duration

	witnessPolicy policy.TLogPolicy
	witnessOpts   WitnessOptions

	mirrorPolicy policy.TLogPolicy
	mirrorOpts   MirroringOptions

	addDecorators []func(AddFn) AddFn
	followers     []Follower

	// garbageCollectionInterval of zero should be interpreted as requesting garbage collection to be disabled.
	garbageCollectionInterval time.Duration

	// shutdownTimeout is the maximum duration to wait for the shutdown process to complete.
	shutdownTimeout time.Duration

	// primarySigner is the "canonical" signer which is used when creating new checkpoints.
	primarySigner note.Signer

	// additionalSigners are additional signers which are used to sign new checkpoints.
	additionalSigners []note.Signer

	origin string
}

// valid returns an error if an invalid combination of options has been set, or nil otherwise.
func (o *AppendOptions) valid(ctx context.Context) error {
	if o.newCP == nil {
		return errors.New("invalid AppendOptions: WithCheckpointSigner must be set")
	}
	if o.primarySigner != nil {
		origin := o.primarySigner.Name()
		for _, signer := range o.additionalSigners {
			if origin != signer.Name() {
				return fmt.Errorf("additional signer name %q does not match primary signer name %q", signer.Name(), origin)
			}
		}
	}
	if o.checkpointRepublishInterval > 0 && o.checkpointRepublishInterval < o.checkpointInterval {
		return fmt.Errorf("invalid AppendOptions: WithCheckpointRepublishInterval (%d) is smaller than WithCheckpointInterval (%d)", o.checkpointRepublishInterval, o.checkpointInterval)
	}
	// Warning: The following warning log will be removed and replaced with returning an error.
	if o.checkpointPublicationTimeout < o.witnessOpts.Timeout {
		slog.WarnContext(ctx, "WithCheckpointPublicationTimeout is smaller than WithWitnessTimeout. This will be updated to a fatal error in the future.",
			slog.Duration("checkpointPublicationTimeout", o.checkpointPublicationTimeout),
			slog.Duration("witnessTimeout", o.witnessOpts.Timeout),
		)
	}
	if o.checkpointPublicationTimeout < o.mirrorOpts.Timeout {
		slog.WarnContext(ctx, "WithCheckpointPublicationTimeout is smaller than WithMirrorTimeout. This will be updated to a fatal error in the future.",
			slog.Duration("checkpointPublicationTimeout", o.checkpointPublicationTimeout),
			slog.Duration("mirrorTimeout", o.mirrorOpts.Timeout),
		)
	}
	if target := max(o.checkpointPublicationTimeout, o.witnessOpts.Timeout, o.mirrorOpts.Timeout); target != o.checkpointPublicationTimeout {
		slog.WarnContext(ctx, "Setting checkpointPublicationTimeout to new value",
			slog.Duration("from", o.checkpointPublicationTimeout),
			slog.Duration("to", target),
		)
		o.checkpointPublicationTimeout = target
	}
	return nil
}

// WithOrigin configures the log's origin.
//
// Most logs SHOULD NOT set this option, but instead rely on the default
// Tessera behaviour of using the signer key's name as the log origin in
// WithCheckpointSinger().
//
// SPEC: https://c2sp.org/tlog-checkpoint
// "The log’s key name in its signature line SHOULD match the origin line."
func (o *AppendOptions) WithOrigin(origin string) *AppendOptions {
	o.origin = origin
	return o
}

// WithAntispam configures the appender to use the antispam mechanism to reduce the number of duplicates which
// can be added to the log.
//
// As a starting point, the minimum size of the of in-memory cache should be set to the configured PushbackThreshold
// of the provided antispam implementation, multiplied by the number of concurrent front-end instances which
// are accepting write-traffic. Data stored in the in-memory cache is relatively small (32 bytes hash, 8 bytes index),
// so we recommend erring on the larger side as there is little downside to over-sizing the cache; consider using
// the DefaultAntispamInMemorySize as the value here.
//
// For more details on how the antispam mechanism works, including tuning guidance, see docs/design/antispam.md.
func (o *AppendOptions) WithAntispam(inMemEntries uint, as Antispam) *AppendOptions {
	o.addDecorators = append(o.addDecorators, newInMemoryDedup(inMemEntries))
	if as != nil {
		o.addDecorators = append(o.addDecorators, as.Decorator())
		o.followers = append(o.followers, as.Follower(o.bundleIDHasher))
	}
	return o
}

// CheckpointPublisher should not be used.
// Deprecated: Use CheckpointPublisherContext.
func (o AppendOptions) CheckpointPublisher(lr LogReader, httpClient *http.Client) func(context.Context, uint64, []byte) ([]byte, error) {
	panic("CheckpointPublisher is deprecated, Tessera should be using CheckpointPublisherContext instead.\nPlease file an issue if this has affected you.")
}

// CheckpointPublisherContext returns a function which should be used to create, sign, and potentially witness and/or mirror a new checkpoint.
func (o AppendOptions) CheckpointPublisherContext(ctx context.Context, lr LogReader, httpClient *http.Client) (func(context.Context, uint64, []byte) ([]byte, error), error) {
	mirrorGateway, err := o.mirrorGateway(ctx, lr, httpClient)
	if err != nil {
		return nil, fmt.Errorf("failed to create mirror gateway: %w", err)
	}

	witnessGateway, err := o.witnessGateway(ctx, lr, httpClient)
	if err != nil {
		return nil, fmt.Errorf("failed to create witness gateway: %w", err)
	}

	// Now return the actual publisher func.
	return func(ctx context.Context, cpSize 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, cpSize, root)
			if err != nil {
				return nil, fmt.Errorf("newCP: %v", err)
			}
			span.AddEvent("Created CP")
			appenderSignedSize.Record(ctx, otel.Clamp64(cpSize))

			var ws, ms []byte
			eg := errgroup.Group{}
			if witnessGateway != nil {
				eg.Go(func() error {
					ctx, cancel := context.WithTimeout(ctx, o.witnessOpts.Timeout)
					defer cancel()

					var err error
					ws, err = witnessCheckpoint(ctx, witnessGateway.CosignCheckpoint, o.witnessPolicy, cp, cpSize, o.witnessOpts.FailOpen, o.witnessOpts.Greedy)
					return err
				})
			}

			if mirrorGateway != nil {
				eg.Go(func() error {
					ctx, cancel := context.WithTimeout(ctx, o.mirrorOpts.Timeout)
					defer cancel()

					var err error
					ms, err = mirrorCheckpoint(ctx, mirrorGateway.CosignCheckpoint, o.mirrorPolicy, cp, cpSize, o.mirrorOpts.FailOpen, false)
					return err
				})
			}

			if err := eg.Wait(); err != nil {
				return nil, fmt.Errorf("failed to fetch cosignatures: %v", err)
			}

			cp = append(append(slices.Clone(cp), ws...), ms...)
			return cp, nil
		})
	}, nil
}

// witnessGateway creates and returns a witnessGateway instance, or nil if no witnesses are configured.
func (o AppendOptions) witnessGateway(ctx context.Context, lr LogReader, httpClient *http.Client) (*witness.WitnessGateway, error) {
	witnesses := []witness.Witness{}
	for _, w := range o.witnessPolicy.Witnesses {
		if w.URL == nil {
			return nil, fmt.Errorf("invalid witness policy: witness %q has no URL", w.Name)
		}
		if w.Verifier == nil {
			return nil, fmt.Errorf("invalid witness policy: witness %q verifier is nil", w.Name)
		}
		witnesses = append(witnesses, witness.Witness{
			URL:       w.URL,
			Verifiers: []note.Verifier{w.Verifier},
		})
	}
	if len(witnesses) == 0 {
		if o.witnessPolicy.Quorum != "" && o.witnessPolicy.Quorum != "none" {
			return nil, fmt.Errorf("invalid witness policy: invalid quorum %q for zero witnesses", o.witnessPolicy.Quorum)
		}
		return nil, nil
	}
	witnessGateway, err := witness.NewGateway(ctx, witness.Options{
		Witnesses:  witnesses,
		HTTPClient: httpClient,
		FetchTiles: lr.ReadTile,
	})
	return witnessGateway, err
}

// mirrorGateway creates and returns a mirrorGateway instance, or nil if no mirrors are configured.
func (o AppendOptions) mirrorGateway(ctx context.Context, lr LogReader, httpClient *http.Client) (*m_gateway.Gateway, error) {
	mirrorURLs := []*url.URL{}
	seen := make(map[string]bool)
	for _, m := range o.mirrorPolicy.Witnesses {
		if m.URL == nil {
			return nil, fmt.Errorf("invalid mirror policy: mirror %q has no URL", m.Name)
		}
		uStr := m.URL.String()
		if !seen[uStr] {
			seen[uStr] = true
			mirrorURLs = append(mirrorURLs, m.URL)
		}
	}
	if len(mirrorURLs) == 0 {
		return nil, nil
	}

	mirrorGateway, err := m_gateway.NewGateway(ctx, m_gateway.Options{
		Mirrors:    mirrorURLs,
		LogReader:  lr,
		HTTPClient: httpClient,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to create gateway: %w", err)
	}
	return mirrorGateway, nil
}

// 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, cosign cosigSource, wPol policy.TLogPolicy, cp []byte, cpSize uint64, failOpen bool, greedy bool) ([]byte, error) {
	return otel.Trace(ctx, "tessera.CheckpointPublisher.Witness", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
		start := time.Now()
		witAttr := []attribute.KeyValue{}

		sigs, err := gatherCosignatures(ctx, "witness", cosign, wPol, cp, cpSize, failOpen, greedy)
		d := time.Since(start)
		if err != nil {
			if !errors.Is(err, errFailedOpen) {
				witAttr = append(witAttr, attribute.String("error.type", "failed"))
				appenderWitnessRequests.Add(ctx, 1, metric.WithAttributes(witAttr...))
				appenderWitnessHistogram.Record(ctx, d.Milliseconds(), metric.WithAttributes(witAttr...))
				slog.WarnContext(ctx, "Failed to collect witness signatures", slog.Any("error", err))
				return nil, err
			}
			witAttr = append(witAttr, attribute.String("error.type", "failed-open"))
		}
		appenderWitnessRequests.Add(ctx, 1, metric.WithAttributes(witAttr...))
		appenderWitnessedSize.Record(ctx, otel.Clamp64(cpSize))
		appenderWitnessHistogram.Record(ctx, d.Milliseconds(), metric.WithAttributes(witAttr...))
		return sigs, nil
	})
}

// 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, cosign cosigSource, mPol policy.TLogPolicy, cp []byte, cpSize uint64, failOpen bool, greedy bool) ([]byte, error) {
	return otel.Trace(ctx, "tessera.CheckpointPublisher.Mirror", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
		sigs, err := gatherCosignatures(ctx, "mirror", cosign, mPol, cp, cpSize, failOpen, greedy)
		if err != nil {
			if !errors.Is(err, errFailedOpen) {
				slog.WarnContext(ctx, "Failed to collect mirror signatures", slog.Any("error", err))
				return nil, err
			}
		}
		return sigs, nil
	})
}

// cosigSource defines a function that can be called to fetch cosignatures.
// Implementations should send cosignatures via the returned channel as they become available, and MUST
// close the channel once no further signatures will be sent, or the context is cancelled.
type cosigSource func(ctx context.Context, cp []byte, cpSize uint64) <-chan []byte

// errFailedOpen is returned by gatherCosignatures if it did not get sufficient cosignatures to satisfy
// the provided policy, but failOpen was true.
var errFailedOpen = errors.New("failed-open")

// gatherCosignatures gathers signatures from a source, applying a policy to determine if the signatures are sufficient.
// It returns a set of signatures which satisfy the policy (potentially more than required if greedy is true), or an error if the policy is not met and failOpen is false.
func gatherCosignatures(ctx context.Context, name string, fetcher cosigSource, pol policy.TLogPolicy, cp []byte, cpSize uint64, failOpen bool, greedy bool) ([]byte, error) {
	// checkPolicy checks if the provided signatures satisfy the given policy.
	checkPolicy := func(sigs []byte, failOpen bool) ([]byte, error) {
		newCP := append(slices.Clone(cp), sigs...)
		if pol.Satisfied(newCP) {
			return sigs, nil
		}
		if failOpen {
			slog.WarnContext(ctx, "policy not met, failing-open", slog.String("name", name))
			return sigs, errFailedOpen
		}
		slog.WarnContext(ctx, "policy not met", slog.String("name", name))
		return sigs, fmt.Errorf("policy not met")
	}

	// collectSigs attempts to collect cosignatures from the provided channel until the policy is met
	// or the context is done.
	collectSigs := func(ctx context.Context, sigCh <-chan []byte) ([]byte, error) {
		var sigBlock bytes.Buffer
		for {
			select {
			case <-ctx.Done():
				sigs, pErr := checkPolicy(sigBlock.Bytes(), failOpen)
				if pErr != nil {
					pErr = fmt.Errorf("%w: timed-out waiting for signatures", pErr)
				}
				return sigs, pErr
			case sig, ok := <-sigCh:
				if !ok {
					// The source has closed the channel, no more signatures will be coming.
					// So check what we have against the policy, and return accordingly.
					sigs, pErr := checkPolicy(sigBlock.Bytes(), failOpen)
					if pErr != nil {
						pErr = fmt.Errorf("%w: no more signatures available", pErr)
					}
					return sigs, pErr
				}

				sigBlock.Write(sig)
				// If we're greedy, we need to keep collecting until we've got all the responses
				// (or the context is cancelled).
				// Otherwise we can return as soon as we've met the policy.
				if !greedy {
					// Don't allow failOpen here, or we'll break out of the collection loop prematurely.
					sigs, err := checkPolicy(sigBlock.Bytes(), false)
					if err == nil {
						return sigs, nil
					}
				}
			}
		}
	}

	return otel.Trace(ctx, "tessera.gatherCosignatures", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
		if len(pol.Witnesses) == 0 {
			return nil, nil
		}

		// A policy can name witnesses and yet be satisfied by the empty set of cosignatures,
		// e.g. one whose quorum is "none". There's nothing to wait for in that case, so publish
		// straight away rather than delaying every checkpoint by a witness round-trip (or, if
		// the witnesses are unreachable, by the full timeout).
		//
		// Greedy is the exception: there we've been explicitly asked to collect whatever surplus
		// cosignatures we can within the time available.
		if !greedy && pol.Satisfied(cp) {
			span.AddEvent("Policy satisfied with no cosignatures")
			return nil, nil
		}

		span.AddEvent("Starting gathering")
		sigCh := fetcher(ctx, cp, cpSize)

		return collectSigs(ctx, sigCh)
	})
}

func (o AppendOptions) BatchMaxAge() time.Duration {
	return o.batchMaxAge
}

func (o AppendOptions) BatchMaxSize() uint {
	return o.batchMaxSize
}

func (o AppendOptions) PushbackMaxOutstanding() uint {
	return o.pushbackMaxOutstanding
}

func (o AppendOptions) EntriesPath() func(uint64, uint8) string {
	return o.entriesPath
}

func (o AppendOptions) CheckpointInterval() time.Duration {
	return o.checkpointInterval
}

func (o AppendOptions) CheckpointRepublishInterval() time.Duration {
	return o.checkpointRepublishInterval
}

func (o AppendOptions) CheckpointPublicationTimeout() time.Duration {
	return o.checkpointPublicationTimeout
}

func (o AppendOptions) GarbageCollectionInterval() time.Duration {
	return o.garbageCollectionInterval
}

func (o AppendOptions) ShutdownTimeout() time.Duration {
	return o.shutdownTimeout
}

// WithCheckpointSigner is an option for setting the note signer and verifier to use when creating and parsing checkpoints.
// This option is mandatory for creating logs where the checkpoint is signed locally, e.g. in
// the Appender mode. This does not need to be provided where the storage will be used to mirror
// other logs.
//
// A primary signer must be provided:
// - the primary signer is the "canonical" signing identity which should be used when creating new checkpoints.
//
// Zero or more additional signers may also be provided.
// This enables cases like:
//   - a rolling key rotation, where checkpoints are signed by both the old and new keys for some period of time,
//   - using different signature schemes for different audiences, etc.
//
// When providing additional signers, their names MUST be identical to the primary signer name, and this name will be used
// as the checkpoint Origin line. This requirement is enforced during Appender initialization.
//
// Checkpoints signed by these signer(s) will be standard checkpoints as defined by https://c2sp.org/tlog-checkpoint.
//
// If no origin was specified using WithOrigin(), configures the origin with the signer's name.
// SPEC: https://c2sp.org/tlog-checkpoint
// "The log’s key name in its signature line SHOULD match the origin line."
func (o *AppendOptions) WithCheckpointSigner(s note.Signer, additionalSigners ...note.Signer) *AppendOptions {
	if o.origin == "" {
		o.origin = s.Name()
	}
	o.primarySigner = s
	o.additionalSigners = additionalSigners
	o.newCP = func(ctx context.Context, size uint64, hash []byte) ([]byte, error) {
		return otel.Trace(ctx, "tessera.SignCheckpoint", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
			// If we're signing a zero-sized tree, the tlog-checkpoint spec says (via RFC6962) that
			// the root must be SHA256 of the empty string, so we'll enforce that here:
			if size == 0 {
				emptyRoot := rfc6962.DefaultHasher.EmptyRoot()
				hash = emptyRoot[:]
			}
			cpRaw := f_log.Checkpoint{
				Origin: o.origin,
				Size:   size,
				Hash:   hash,
			}.Marshal()

			n, err := note.Sign(&note.Note{Text: string(cpRaw)}, append([]note.Signer{s}, additionalSigners...)...)
			if err != nil {
				return nil, fmt.Errorf("note.Sign: %w", err)
			}

			return n, nil
		})
	}
	return o
}

// WithBatching configures the batching behaviour of leaves being sequenced.
// A batch will be allowed to grow in memory until either:
//   - the number of entries in the batch reach maxSize
//   - the first entry in the batch has reached maxAge
//
// At this point the batch will be sent to the sequencer.
//
// Configuring these parameters allows the personality to tune to get the desired
// balance of sequencing latency with cost. In general, larger batches allow for
// lower cost of operation, where more frequent batches reduce the amount of time
// required for entries to be included in the log.
//
// If this option isn't provided, storage implementations with use the DefaultBatchMaxSize and DefaultBatchMaxAge consts above.
func (o *AppendOptions) WithBatching(maxSize uint, maxAge time.Duration) *AppendOptions {
	o.batchMaxSize = maxSize
	o.batchMaxAge = maxAge
	return o
}

// WithPushback allows configuration of when the storage should start pushing back on add requests.
//
// maxOutstanding is the number of "in-flight" add requests - i.e. the number of entries with sequence numbers
// assigned, but which are not yet integrated into the log.
func (o *AppendOptions) WithPushback(maxOutstanding uint) *AppendOptions {
	o.pushbackMaxOutstanding = maxOutstanding
	return o
}

// WithCheckpointInterval configures the frequency at which Tessera will attempt to create & publish
// new checkpoints.
//
// Well behaved clients of the log will only "see" newly sequenced entries once a new checkpoint is published,
// so it's important to set that value such that it works well with your ecosystem.
//
// Regularly publishing new checkpoints:
//   - helps show that the log is "live", even if no entries are being added.
//   - enables clients of the log to reason about how frequently they need to have their
//     view of the log refreshed, which in turn helps reduce work/load across the ecosystem.
//
// Note that this option probably only makes sense for long-lived applications (e.g. HTTP servers).
//
// If this option isn't provided, storage implementations will use the DefaultCheckpointInterval const above.
func (o *AppendOptions) WithCheckpointInterval(interval time.Duration) *AppendOptions {
	o.checkpointInterval = interval
	return o
}

// WithCheckpointRepublishInterval configures the frequency at which Tessera will allow re-publishing
// checkpoints where the log hasn't grown since the last checkpoint was published.
//
// Setting this less than or equal to zero will disable republication of unchanged checkpoints.
func (o *AppendOptions) WithCheckpointRepublishInterval(interval time.Duration) *AppendOptions {
	o.checkpointRepublishInterval = interval
	return o
}

// WithCheckpointPublicationTimeout configures the timeout used when creating and publishing checkpoints.
func (o *AppendOptions) WithCheckpointPublicationTimeout(timeout time.Duration) *AppendOptions {
	o.checkpointPublicationTimeout = timeout
	return o
}

// WithWitnessPolicy configures the set of witnesses that Tessera will contact in order to cosign
// a checkpoint before publishing it. A request will be sent to every witness referenced by the policy.
// The checkpoint will be accepted for publishing when a sufficient number of witnesses to satisfy
// the policy have responded.
//
// If this method is not called, then witnessing will not be performed.
func (o *AppendOptions) WithWitnessPolicy(witnessPolicy policy.TLogPolicy, opts *WitnessOptions) *AppendOptions {
	if opts == nil {
		opts = &WitnessOptions{}
	}
	if opts.Timeout == 0 {
		opts.Timeout = DefaultWitnessTimeout
	}

	o.witnessPolicy = witnessPolicy
	o.witnessOpts = *opts
	return o
}

// WithWitnesses configures the set of witnesses that Tessera will contact in order to cosign
// a checkpoint before publishing it.
//
// Deprecated: Use WithWitnessPolicy instead.
func (o *AppendOptions) WithWitnesses(witnesses WitnessGroup, opts *WitnessOptions) *AppendOptions {
	p, err := witnesses.toPolicy()
	if err != nil {
		panic(fmt.Sprintf("invalid WitnessGroup: %v", err))
	}
	return o.WithWitnessPolicy(p, opts)
}

// WithMirrorPolicy configures the set of tlog-mirror servers that Tessera will contact in order to obtain
// mirror cosignatures on a checkpoint before publishing it.
//
// Requests will be sent to every mirror referenced by the policy using the tlog-mirror API at the configured URL.
// The checkpoint will be accepted for publishing when a sufficient number of mirrors to satisfy the policy
// have responded.
//
// If this method is not called, then mirroring will not be performed.
func (o *AppendOptions) WithMirrorPolicy(mirrorPolicy policy.TLogPolicy, opts *MirroringOptions) *AppendOptions {
	if opts == nil {
		opts = &MirroringOptions{}
	}
	if opts.Timeout == 0 {
		opts.Timeout = DefaultMirrorTimeout
	}

	o.mirrorPolicy = mirrorPolicy
	o.mirrorOpts = *opts
	return o
}

// WithMirrors configures the set of tlog-mirror servers that Tessera will contact in order to obtain
// mirror cosignatures on a checkpoint before publishing it.
//
// Deprecated: Use WithMirrorPolicy instead.
func (o *AppendOptions) WithMirrors(mirrors WitnessGroup, opts *MirroringOptions) *AppendOptions {
	p, err := mirrors.toPolicy()
	if err != nil {
		panic(fmt.Sprintf("invalid WitnessGroup: %v", err))
	}
	return o.WithMirrorPolicy(p, opts)
}

// WitnessOptions contains extra optional configuration for how Tessera should use/interact with
// a user-provided witness policy.
type WitnessOptions struct {
	// Timeout is the maximum time to wait while attempting to satisfy the configured witness policy.
	//
	// If the policy has not already been satisfied at the point this duration has passed, Tessera
	// will stop waiting for more responses. The FailOpen option below controls whether or not the
	// checkpoint will be published in this case.
	//
	// If unset, uses DefaultWitnessTimeout.
	Timeout time.Duration

	// FailOpen controls whether a checkpoint, for which the witness policy was unable to be met,
	// should still be published.
	//
	// This setting is intended only for facilitating early "non-blocking" adoption of witnessing,
	// and will be disabled and/or removed in the future.
	FailOpen bool

	// Greedy indicates that Tessera should attempt to collect as many cosignatures as possible within the time available.
	// If unset, Tessera will stop waiting for more responses as soon as the policy has been met.
	Greedy bool
}

// MirroringOptions contains extra optional configuration for how Tessera should use/interact with
// the tlog-mirror servers.
type MirroringOptions struct {
	// Timeout is the maximum time to wait while attempting to satisfy the configured mirror policy.
	//
	// If the policy has not already been satisfied at the point this duration has passed, Tessera
	// will stop waiting for more responses. The FailOpen option below controls whether or not the
	// checkpoint will be published in this case.
	//
	// If unset, uses DefaultMirrorTimeout.
	Timeout time.Duration

	// FailOpen controls whether a checkpoint, for which the mirror policy was unable to be met,
	// should still be published.
	//
	// This setting is intended only for facilitating early "non-blocking" adoption of mirroring,
	// and will be disabled and/or removed in the future.
	FailOpen bool
}

// WithGarbageCollectionInterval allows the interval between scans to remove obsolete partial
// tiles and entry bundles.
//
// Setting to zero disables garbage collection.
func (o *AppendOptions) WithGarbageCollectionInterval(interval time.Duration) *AppendOptions {
	o.garbageCollectionInterval = interval
	return o
}

// WithShutdownTimeout allows the maximum duration Tessera will wait for the shutdown
// process to complete. Any delays in checkpoint finalisation or logs which have not caught up
// to the checkpoint will cause a timeout error.
//
// The default value is DefaultShutdownTimeout.
func (o *AppendOptions) WithShutdownTimeout(timeout time.Duration) *AppendOptions {
	o.shutdownTimeout = timeout
	return o
}

// LogValue implements slog.LogValuer to allow safe serialization of AppendOptions.
func (o *AppendOptions) LogValue() slog.Value {
	if o == nil {
		return slog.Value{}
	}

	attrs := []slog.Attr{
		slog.Duration("batchMaxAge", o.batchMaxAge),
		slog.Uint64("batchMaxSize", uint64(o.batchMaxSize)),
		slog.Uint64("pushbackMaxOutstanding", uint64(o.pushbackMaxOutstanding)),
		slog.Uint64("maxEntrySize", uint64(o.maxEntrySize)),
		slog.Duration("checkpointInterval", o.checkpointInterval),
		slog.Duration("checkpointRepublishInterval", o.checkpointRepublishInterval),
		slog.Duration("garbageCollectionInterval", o.garbageCollectionInterval),
		slog.Duration("shutdownTimeout", o.shutdownTimeout),
	}

	if o.primarySigner != nil {
		attrs = append(attrs, slog.String("primarySigner", o.primarySigner.Name()))
	}

	if len(o.additionalSigners) > 0 {
		names := make([]string, len(o.additionalSigners))
		for i, s := range o.additionalSigners {
			names[i] = s.Name()
		}
		attrs = append(attrs, slog.Any("additionalSigners", names))
	}

	if len(o.witnessPolicy.Witnesses) > 0 {
		seen := make(map[string]bool)
		urls := make([]string, 0, len(o.witnessPolicy.Witnesses))
		for _, w := range o.witnessPolicy.Witnesses {
			if w.URL != nil {
				u := w.URL.String()
				if !seen[u] {
					seen[u] = true
					urls = append(urls, u)
				}
			}
		}
		attrs = append(attrs, slog.Group("witnesses",
			slog.Int("groups", len(o.witnessPolicy.Groups)),
			slog.Any("quorum", o.witnessPolicy.Quorum),
			slog.Any("endpoints", urls),
		))
	}
	if len(o.mirrorPolicy.Witnesses) > 0 {
		seen := make(map[string]bool)
		urls := make([]string, 0, len(o.mirrorPolicy.Witnesses))
		for _, w := range o.mirrorPolicy.Witnesses {
			if w.URL != nil {
				u := w.URL.String()
				if !seen[u] {
					seen[u] = true
					urls = append(urls, u)
				}
			}
		}
		attrs = append(attrs, slog.Group("mirrors",
			slog.Int("groups", len(o.mirrorPolicy.Groups)),
			slog.Any("quorum", o.mirrorPolicy.Quorum),
			slog.Any("endpoints", urls),
		))
	}

	if len(o.followers) > 0 {
		names := make([]string, len(o.followers))
		for i, f := range o.followers {
			names[i] = f.Name()
		}
		attrs = append(attrs, slog.Any("followers", names))
	}

	return slog.GroupValue(attrs...)
}