gcp.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 gcp contains a GCP-based storage implementation for Tessera.
//
// TODO: decide whether to rename this package.
//
// This storage implementation uses GCS for long-term storage and serving of
// entry bundles and log tiles, and Spanner for coordinating updates to GCS
// when multiple instances of a personality binary are running.
//
// A single GCS bucket is used to hold entry bundles and log internal tiles.
// The object keys for the bucket are selected so as to conform to the
// expected layout of a tile-based log.
//
// A Spanner database provides a transactional mechanism to allow multiple
// frontends to safely update the contents of the log.
package gcp

import (
	"bytes"
	"context"
	"encoding/gob"
	"errors"
	"fmt"
	"io"
	"math"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"sync"
	"time"

	"cloud.google.com/go/spanner"
	database "cloud.google.com/go/spanner/admin/database/apiv1"
	adminpb "cloud.google.com/go/spanner/admin/database/apiv1/databasepb"
	"cloud.google.com/go/spanner/apiv1/spannerpb"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"

	"log/slog"

	gcs "cloud.google.com/go/storage"
	"github.com/google/go-cmp/cmp"
	"github.com/transparency-dev/merkle/rfc6962"
	"github.com/transparency-dev/tessera"
	"github.com/transparency-dev/tessera/api"
	"github.com/transparency-dev/tessera/api/layout"
	"github.com/transparency-dev/tessera/internal/fetcher"
	"github.com/transparency-dev/tessera/internal/migrate"
	"github.com/transparency-dev/tessera/internal/otel"
	"github.com/transparency-dev/tessera/internal/parse"
	storage "github.com/transparency-dev/tessera/storage/internal"
	"go.opentelemetry.io/otel/trace"
	"golang.org/x/sync/errgroup"
	"google.golang.org/api/googleapi"
	"google.golang.org/api/iterator"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

const (
	// minCheckpointInterval is the shortest permitted interval between updating published checkpoints.
	// GCS has a rate limit 1 update per second for individual objects, but we've observed that attempting
	// to update at exactly that rate still results in the occasional refusal, so bake in a little wiggle
	// room.
	minCheckpointInterval = 1100 * time.Millisecond

	logContType      = "application/octet-stream"
	ckptContType     = "text/plain; charset=utf-8"
	logCacheControl  = "max-age=604800,immutable"
	ckptCacheControl = "no-cache"

	DefaultIntegrationSizeLimit = 5 * 4096

	// defaultSeqTableMaxBatchByteSize is the default maximum byte size of a batch of entries to be written to the
	// "V" field in the "Seq" table. This is set to just under 10 MiB, the maximum size of the Spanner
	// BYTES column, with some headroom for the gob encoding.
	defaultSeqTableMaxBatchByteSize = 9 << 20 // 9 MiB

	// SchemaCompatibilityVersion represents the expected version (e.g. layout & serialisation) of stored data.
	//
	// A binary built with a given version of the Tessera library is compatible with stored data created by a different version
	// of the library if and only if this value is the same as the compatibilityVersion stored in the Tessera table.
	//
	// NOTE: if changing this version, you need to consider whether end-users are going to update their schema instances to be
	// compatible with the new format, and provide a means to do it if so.
	SchemaCompatibilityVersion = 1

	// defaultAssignEntriesTimeout is the default context timeout applied when assigning a batch of entries to the Spanner sequencer.
	defaultAssignEntriesTimeout = 2 * time.Second
	// defaultIntegrationTimeout is the default context timeout applied when undertaking an integration task.
	defaultIntegrationTimeout = 10 * time.Second
	// defaultGCTimeout is the default context timeout applied when undertaking a garbage collection task.
	defaultGCTimeout = 30 * time.Second
)

// Storage is a GCP based storage implementation for Tessera.
type Storage struct {
	cfg Config
}

// sequencer describes a type which knows how to sequence entries.
//
// TODO(al): rename this as it's really more of a coordination for the log.
type sequencer interface {
	// assignEntries should durably allocate contiguous index numbers to the provided entries.
	assignEntries(ctx context.Context, entries []*tessera.Entry) error
	// consumeEntries should call the provided function with up to limit previously sequenced entries.
	// If the call to consumeFunc returns no error, the entries should be considered to have been consumed.
	// If any entries were successfully consumed, the implementation should also return true; this
	// serves as a weak hint that there may be more entries to be consumed.
	// If forceUpdate is true, then the consumeFunc should be called, with an empty slice of entries if
	// necessary. This allows the log self-initialise in a transactionally safe manner.
	consumeEntries(ctx context.Context, limit uint64, f consumeFunc, forceUpdate bool) (bool, error)
	// currentTree returns the tree state of the currently integrated tree according to the IntCoord table.
	currentTree(ctx context.Context) (uint64, []byte, error)
	// nextIndex returns the next available index in the log.
	nextIndex(ctx context.Context) (uint64, error)
	// publishCheckpoint coordinates the publication of new checkpoints based on the current integrated tree.
	publishCheckpoint(ctx context.Context, minStaleActive, minStaleRepub time.Duration, f func(ctx context.Context, size uint64, root []byte) error) (time.Time, error)
	// garbageCollect coordinates the removal of unneeded partial tiles/entry bundles for the provided tree size, up to a maximum number of deletes per invocation.
	garbageCollect(ctx context.Context, treeSize uint64, maxDeletes uint, removePrefix func(ctx context.Context, prefix string) error, entriesPath func(uint64, uint8) string) error
}

// consumeFunc is the signature of a function which can consume entries from the sequencer and integrate
// them into the log.
// Returns the new rootHash once all passed entries have been integrated.
type consumeFunc func(ctx context.Context, from uint64, entries []storage.SequencedEntry) ([]byte, error)

// Config holds GCP project and resource configuration for a storage instance.
type Config struct {
	// GCSClient will be  used to interact with GCS. If unset, Tessera will create one.
	GCSClient *gcs.Client
	// SpannerClient will be used to interact with Spanner. If unset, Tessera will create one.
	SpannerClient *spanner.Client
	// HTTPClient will be used for other HTTP requests. If unset, Tessera will use the net/http DefaultClient.
	HTTPClient *http.Client

	// Bucket is the name of the GCS bucket to use for storing log state.
	Bucket string
	// BucketPrefix is an optional prefix to prepend to all log resource paths.
	// This can be used e.g. to store multiple logs in the same bucket.
	BucketPrefix string
	// Spanner is the GCP resource URI of the spanner database instance to use.
	Spanner string
	// SpannerTablePrefix is an optional prefix to prepend to the names of all Spanner tables.
	// If set, it must start with a letter, contain only letters, digits, or underscores
	// (e.g. "log1_"), and be at most 64 characters long.
	// It's recommended to derive this prefix from the log's origin string, to
	// make it easy to associate tables with specific logs.
	//
	// TODO: consider providing a mechanism to set both BucketPrefix and SpannerTablePrefix
	// from a single string to avoid misconfiguration foot-guns.
	SpannerTablePrefix string
}

// tablePrefixRE matches valid values for a Spanner table prefix: empty, or a leading
// letter followed by up to 63 letters, digits, or underscores.
var tablePrefixRE = regexp.MustCompile(`^([A-Za-z][A-Za-z0-9_]{0,63})?$`)

// New creates a new instance of the GCP based Storage.
func New(ctx context.Context, cfg Config) (tessera.Driver, error) {
	if !tablePrefixRE.MatchString(cfg.SpannerTablePrefix) {
		return nil, fmt.Errorf("invalid SpannerTablePrefix %q: must start with a letter, contain only letters, digits, or underscores, and be at most 64 characters long", cfg.SpannerTablePrefix)
	}
	if cfg.HTTPClient == nil {
		cfg.HTTPClient = http.DefaultClient
	}
	return &Storage{
		cfg: cfg,
	}, nil
}

type LogReader struct {
	lrs            logResourceStore
	integratedSize func(context.Context) (uint64, error)
	nextIndex      func(context.Context) (uint64, error)
}

func (lr *LogReader) ReadCheckpoint(ctx context.Context) ([]byte, error) {
	return otel.Trace(ctx, "tessera.storage.gcp.ReadCheckpoint", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
		r, err := lr.lrs.getCheckpoint(ctx)
		if err != nil {
			if errors.Is(err, gcs.ErrObjectNotExist) {
				return r, os.ErrNotExist
			}
		}
		return r, err
	})
}

func (lr *LogReader) ReadTile(ctx context.Context, l, i uint64, p uint8) ([]byte, error) {
	return otel.Trace(ctx, "tessera.storage.gcp.ReadTile", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
		return fetcher.PartialOrFullResource(ctx, p, func(ctx context.Context, p uint8) ([]byte, error) {
			return lr.lrs.getTile(ctx, l, i, p)
		})
	})
}

func (lr *LogReader) ReadEntryBundle(ctx context.Context, i uint64, p uint8) ([]byte, error) {
	return otel.Trace(ctx, "tessera.storage.gcp.ReadEntryBundle", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
		return fetcher.PartialOrFullResource(ctx, p, func(ctx context.Context, p uint8) ([]byte, error) {
			return lr.lrs.getEntryBundle(ctx, i, p)
		})
	})
}

func (lr *LogReader) IntegratedSize(ctx context.Context) (uint64, error) {
	return otel.Trace(ctx, "tessera.storage.gcp.IntegratedSize", tracer, func(ctx context.Context, span trace.Span) (uint64, error) {
		return lr.integratedSize(ctx)
	})
}

func (lr *LogReader) NextIndex(ctx context.Context) (uint64, error) {
	return otel.Trace(ctx, "tessera.storage.gcp.NextIndex", tracer, func(ctx context.Context, span trace.Span) (uint64, error) {
		return lr.nextIndex(ctx)
	})
}

// Appender creates a new tessera.Appender lifecycle object.
func (s *Storage) Appender(ctx context.Context, opts *tessera.AppendOptions) (*tessera.Appender, tessera.LogReader, error) {
	if s.cfg.GCSClient == nil {
		var err error
		s.cfg.GCSClient, err = gcs.NewClient(ctx, gcs.WithJSONReads())
		if err != nil {
			return nil, nil, fmt.Errorf("failed to create GCS client: %v", err)
		}
	}
	gs := &gcsStorage{
		gcsClient:    s.cfg.GCSClient,
		bucket:       s.cfg.Bucket,
		bucketPrefix: s.cfg.BucketPrefix,
	}

	var err error
	if s.cfg.SpannerClient == nil {
		s.cfg.SpannerClient, err = spanner.NewClient(ctx, s.cfg.Spanner)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to connect to Spanner: %v", err)
		}
	}
	table := func(t string) string {
		return s.cfg.SpannerTablePrefix + t
	}
	if err := initDB(ctx, s.cfg.Spanner, table); err != nil {
		return nil, nil, fmt.Errorf("failed to verify/init Spanner schema: %v", err)
	}

	seq, err := newSpannerCoordinator(ctx, s.cfg.SpannerClient, table, uint64(opts.PushbackMaxOutstanding()))
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create Spanner coordinator: %v", err)
	}

	a, lr, err := s.newAppender(ctx, gs, seq, opts)
	if err != nil {
		return nil, nil, err
	}
	return &tessera.Appender{
		Add: a.Add,
	}, lr, nil
}

// newAppender creates and initialises a tessera.Appender struct with the provided underlying storage implementations.
func (s *Storage) newAppender(ctx context.Context, o objStore, seq *spannerCoordinator, opts *tessera.AppendOptions) (*Appender, tessera.LogReader, error) {
	if opts.CheckpointInterval() < minCheckpointInterval {
		return nil, nil, fmt.Errorf("requested CheckpointInterval (%v) is less than minimum permitted %v", opts.CheckpointInterval(), minCheckpointInterval)
	}

	a := &Appender{
		logStore: &logResourceStore{
			objStore:    o,
			entriesPath: opts.EntriesPath(),
		},
		sequencer:       seq,
		cpUpdated:       make(chan struct{}),
		entriesAssigned: make(chan struct{}, 1),
	}
	a.queue = storage.NewQueue(ctx, opts.BatchMaxAge(), opts.BatchMaxSize(), func(ctx context.Context, entries []*tessera.Entry) error {
		ctx, cancel := context.WithTimeout(ctx, defaultAssignEntriesTimeout)
		defer cancel()
		if err := a.sequencer.assignEntries(ctx, entries); err != nil {
			return err
		}
		select {
		case a.entriesAssigned <- struct{}{}:
		default:
		}
		return nil
	})

	reader := &LogReader{
		lrs: *a.logStore,
		integratedSize: func(ctx context.Context) (uint64, error) {
			s, _, err := a.sequencer.currentTree(ctx)
			return s, err
		},
		nextIndex: a.sequencer.nextIndex,
	}
	var err error
	a.newCP, err = opts.CheckpointPublisherContext(ctx, reader, s.cfg.HTTPClient)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create checkpoint publisher: %v", err)
	}

	if err := a.init(ctx); err != nil {
		return nil, nil, fmt.Errorf("failed to initialise log storage: %v", err)
	}

	go a.integrateEntriesJob(ctx)
	go a.publishCheckpointJob(ctx, opts.CheckpointInterval(), opts.CheckpointRepublishInterval(), opts.CheckpointPublicationTimeout())
	if i := opts.GarbageCollectionInterval(); i > 0 {
		go a.garbageCollectorJob(ctx, i)
	}

	return a, reader, nil
}

// Appender is an implementation of the Tessera appender lifecycle contract.
type Appender struct {
	newCP func(context.Context, uint64, []byte) ([]byte, error)

	sequencer sequencer
	logStore  *logResourceStore

	queue *storage.Queue

	cpUpdated       chan struct{}
	entriesAssigned chan struct{}
}

// Add is the entrypoint for adding entries to a sequencing log.
func (a *Appender) Add(ctx context.Context, e *tessera.Entry) tessera.IndexFuture {
	ctx, span := tracer.Start(ctx, "tessera.storage.gcp.Add")
	defer span.End()

	// Reject entries which are too large to fit in the Seq table even as a single-entry batch.
	if size := len(e.Data()); size > defaultSeqTableMaxBatchByteSize {
		return func() (tessera.Index, error) {
			return tessera.Index{}, fmt.Errorf("entry size %d exceeds maximum allowed %d", size, defaultSeqTableMaxBatchByteSize)
		}
	}

	return a.queue.Add(ctx, e)
}

// integrateEntriesJob periodically append newly sequenced entries.
//
// Blocks until ctx is done.
func (a *Appender) integrateEntriesJob(ctx context.Context) {
	t := time.NewTicker(1 * time.Second)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-a.entriesAssigned:
		case <-t.C:
		}

		if err := otel.TraceErr(ctx, "tessera.storage.gcp.integrateEntriesJob", tracer, func(ctx context.Context, span trace.Span) error {
			start := time.Now()
			defer func() {
				opsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("integrateEntries")))
			}()

			ctx, cancel := context.WithTimeout(ctx, defaultIntegrationTimeout)
			defer cancel() // Note: ok because we're in a func passed to TraceErr here!

			workDone, err := a.sequencer.consumeEntries(ctx, DefaultIntegrationSizeLimit, a.integrateEntries, false)
			if err != nil {
				return fmt.Errorf("integrateEntriesJob: %v", err)
			}
			if workDone {
				select {
				case a.cpUpdated <- struct{}{}:
				default:
				}
			}
			return nil
		}, trace.WithAttributes(otel.PeriodicKey.Bool(true))); err != nil {
			slog.ErrorContext(ctx, "integrateEntriesJob failed", slog.Any("error", err))
		}
	}
}

// publishCheckpointJob periodically attempts to publish a new checkpoint representing the current state
// of the tree, once per interval.
//
// Blocks until ctx is done.
func (a *Appender) publishCheckpointJob(ctx context.Context, pubInterval, republishInterval, publicationTimeout time.Duration) {
	t := time.NewTicker(pubInterval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-a.cpUpdated:
		case <-t.C:
		}
		if err := otel.TraceErr(ctx, "tessera.storage.gcp.publishCheckpointJob", tracer, func(ctx context.Context, span trace.Span) error {

			ctx, cancel := context.WithTimeout(ctx, publicationTimeout)
			defer cancel() // Note: ok because we're in a func passed to TraceErr here!

			nextPub, err := a.sequencer.publishCheckpoint(ctx, pubInterval, republishInterval, a.updateCheckpoint)
			if err != nil {
				return fmt.Errorf("publishCheckpoint failed: %v", err)
			}
			// Schedule a checkpoint update immediately, if an updated is due.
			t.Reset(max(time.Millisecond, time.Until(nextPub)))
			return nil
		}, trace.WithAttributes(otel.PeriodicKey.Bool(true))); err != nil {
			t.Reset(pubInterval)
			slog.ErrorContext(ctx, "publishCheckpoint failed", slog.Any("error", err))
		}
	}
}

// garbageCollectorJob is a long-running function which handles the removal of obsolete partial tiles
// and entry bundles.
// Blocks until ctx is done.
func (a *Appender) garbageCollectorJob(ctx context.Context, i time.Duration) {
	t := time.NewTicker(i)
	defer t.Stop()

	// Entirely arbitrary number.
	maxBundlesPerRun := uint(100)

	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
		if err := otel.TraceErr(ctx, "tessera.storage.gcp.garbageCollectJob", tracer, func(ctx context.Context, span trace.Span) error {
			ctx, cancel := context.WithTimeout(ctx, defaultGCTimeout)
			defer cancel() // Note: ok because we're in a func passed to TraceErr here!

			// Figure out the size of the latest published checkpoint - we can't be removing partial tiles implied by
			// that checkpoint just because we've done an integration and know about a larger (but as yet unpublished)
			// checkpoint!
			cp, err := a.logStore.getCheckpoint(ctx)
			if err != nil {
				return fmt.Errorf("failed to get published checkpoint: %v", err)
			}
			_, pubSize, _, err := parse.CheckpointUnsafe(cp)
			if err != nil {
				return fmt.Errorf("failed to parse published checkpoint: %v", err)
			}

			if err := a.sequencer.garbageCollect(ctx, pubSize, maxBundlesPerRun, a.logStore.objStore.deleteObjectsWithPrefix, a.logStore.entriesPath); err != nil {
				return fmt.Errorf("garbageCollect failed: %v", err)
			}
			return nil
		}, trace.WithAttributes(otel.PeriodicKey.Bool(true))); err != nil {
			slog.WarnContext(ctx, "garbageCollectTask failed", slog.Any("error", err))
		}
	}

}

// init ensures that the storage represents a log in a valid state.
func (a *Appender) init(ctx context.Context) error {
	if _, err := a.logStore.getCheckpoint(ctx); err != nil {
		if errors.Is(err, gcs.ErrObjectNotExist) {
			// No checkpoint exists, do a forced (possibly empty) integration to create one in a safe
			// way (setting the checkpoint directly here would not be safe as it's outside the transactional
			// framework which prevents the tree from rolling backwards or otherwise forking).
			ctx, c := context.WithTimeout(ctx, defaultIntegrationTimeout)
			defer c()
			if _, err := a.sequencer.consumeEntries(ctx, DefaultIntegrationSizeLimit, a.integrateEntries, true); err != nil {
				return fmt.Errorf("forced integrate: %v", err)
			}
			select {
			case a.cpUpdated <- struct{}{}:
			default:
			}
			return nil
		}
		return fmt.Errorf("failed to read checkpoint: %v", err)
	}

	return nil
}

func (a *Appender) updateCheckpoint(ctx context.Context, size uint64, root []byte) error {
	return otel.TraceErr(ctx, "tessera.storage.gcp.updateCheckpoint", tracer, func(ctx context.Context, span trace.Span) error {
		span.SetAttributes(treeSizeKey.Int64(otel.Clamp64(size)))

		cpRaw, err := a.newCP(ctx, size, root)
		if err != nil {
			return fmt.Errorf("newCP: %v", err)
		}

		if err := a.logStore.setCheckpoint(ctx, cpRaw); err != nil {
			return fmt.Errorf("writeCheckpoint: %v", err)
		}

		slog.DebugContext(ctx, "Created and stored latest checkpoint", slog.Uint64("size", size), slog.String("root", fmt.Sprintf("%x", root)))

		return nil
	})
}

// objStore describes a type which can store and retrieve objects.
type objStore interface {
	getObject(ctx context.Context, obj string) ([]byte, *gcs.ReaderObjectAttrs, error)
	setObject(ctx context.Context, obj string, data []byte, cond *gcs.Conditions, contType string, cacheCtl string) error
	deleteObjectsWithPrefix(ctx context.Context, prefix string) error
}

// logResourceStore knows how to read and write entries which represent a tiles log inside an objStore.
type logResourceStore struct {
	objStore    objStore
	entriesPath func(uint64, uint8) string
}

func (lrs *logResourceStore) setCheckpoint(ctx context.Context, cpRaw []byte) error {
	return lrs.objStore.setObject(ctx, layout.CheckpointPath, cpRaw, nil, ckptContType, ckptCacheControl)
}

func (lrs *logResourceStore) getCheckpoint(ctx context.Context) ([]byte, error) {
	r, attr, err := lrs.objStore.getObject(ctx, layout.CheckpointPath)
	if err != nil {
		return nil, err
	}

	checkpointAgeHistogram.Record(ctx, time.Since(attr.LastModified).Milliseconds())
	return r, err
}

// setTile idempotently stores the provided tile at the location implied by the given level, index, and treeSize.
//
// The location to which the tile is written is defined by the tile layout spec.
func (s *logResourceStore) setTile(ctx context.Context, level, index uint64, partial uint8, data []byte) error {
	start := time.Now()

	tPath := layout.TilePath(level, index, partial)
	err := s.objStore.setObject(ctx, tPath, data, &gcs.Conditions{DoesNotExist: true}, logContType, logCacheControl)
	opsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("writeTile")))
	return err
}

// getTile retrieves the raw tile from the provided location.
//
// The location to which the tile is written is defined by the tile layout spec.
func (s *logResourceStore) getTile(ctx context.Context, level, index uint64, partial uint8) ([]byte, error) {
	start := time.Now()

	tPath := layout.TilePath(level, index, partial)
	d, _, err := s.objStore.getObject(ctx, tPath)
	opsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("readTile")))
	return d, err
}

// getTiles returns the tiles with the given tile-coords for the specified log size.
//
// Tiles are returned in the same order as they're requested, nils represent tiles which were not found.
func (s *logResourceStore) getTiles(ctx context.Context, tileIDs []storage.TileID, logSize uint64) ([]*api.HashTile, error) {
	return otel.Trace(ctx, "tessera.storage.gcp.getTiles", tracer, func(ctx context.Context, span trace.Span) ([]*api.HashTile, error) {
		r := make([]*api.HashTile, len(tileIDs))
		errG := errgroup.Group{}
		for i, id := range tileIDs {
			i := i
			id := id
			errG.Go(func() error {
				objName := layout.TilePath(id.Level, id.Index, layout.PartialTileSize(id.Level, id.Index, logSize))
				data, _, err := s.objStore.getObject(ctx, objName)
				if err != nil {
					if errors.Is(err, gcs.ErrObjectNotExist) {
						// Depending on context, this may be ok.
						// We'll signal to higher levels that it wasn't found by returning a nil for this tile.
						return nil
					}
					return err
				}
				t := &api.HashTile{}
				if err := t.UnmarshalText(data); err != nil {
					return fmt.Errorf("unmarshal(%q): %v", objName, err)
				}
				r[i] = t
				return nil
			})
		}
		if err := errG.Wait(); err != nil {
			return nil, err
		}
		return r, nil
	})
}

// getEntryBundle returns the serialised entry bundle at the location described by the given index and partial size.
// A partial size of zero implies a full tile.
//
// Returns a wrapped os.ErrNotExist if the bundle does not exist.
func (s *logResourceStore) getEntryBundle(ctx context.Context, bundleIndex uint64, p uint8) ([]byte, error) {
	objName := s.entriesPath(bundleIndex, p)
	data, _, err := s.objStore.getObject(ctx, objName)
	if err != nil {
		if errors.Is(err, gcs.ErrObjectNotExist) {
			// Return the generic NotExist error so that higher levels can differentiate
			// between this and other errors.
			return nil, fmt.Errorf("%v: %w", objName, os.ErrNotExist)
		}
		return nil, err
	}

	return data, nil
}

// setEntryBundle idempotently stores the serialised entry bundle at the location implied by the bundleIndex and treeSize.
func (s *logResourceStore) setEntryBundle(ctx context.Context, bundleIndex uint64, p uint8, bundleRaw []byte) error {
	objName := s.entriesPath(bundleIndex, p)
	// Note that setObject does an idempotent interpretation of DoesNotExist - it only
	// returns an error if the named object exists _and_ contains different data to what's
	// passed in here.
	if err := s.objStore.setObject(ctx, objName, bundleRaw, &gcs.Conditions{DoesNotExist: true}, logContType, logCacheControl); err != nil {
		return fmt.Errorf("setObject(%q): %v", objName, err)

	}
	return nil
}

// integrateEntries appends the provided entries into the log starting at fromSeq.
//
// Returns the new root hash of the log with the entries added.
func (a *Appender) integrateEntries(ctx context.Context, fromSeq uint64, entries []storage.SequencedEntry) ([]byte, error) {
	return otel.Trace(ctx, "tessera.storage.gcp.integrateEntries", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
		var newRoot []byte

		errG := errgroup.Group{}

		errG.Go(func() error {
			if err := a.updateEntryBundles(ctx, fromSeq, entries); err != nil {
				return fmt.Errorf("updateEntryBundles: %v", err)
			}
			return nil
		})

		errG.Go(func() error {
			lh := make([][]byte, len(entries))
			for i, e := range entries {
				lh[i] = e.LeafHash
			}
			r, err := integrate(ctx, fromSeq, lh, a.logStore)
			if err != nil {
				return fmt.Errorf("integrate: %v", err)
			}
			newRoot = r
			return nil
		})
		if err := errG.Wait(); err != nil {
			return nil, err
		}
		return newRoot, nil
	})
}

// integrate adds the provided leaf hashes to the merkle tree, starting at the provided location.
func integrate(ctx context.Context, fromSeq uint64, lh [][]byte, logStore *logResourceStore) ([]byte, error) {
	return otel.Trace(ctx, "tessera.storage.gcp.integrate", tracer, func(ctx context.Context, span trace.Span) ([]byte, error) {
		span.SetAttributes(fromSizeKey.Int64(otel.Clamp64(fromSeq)), numEntriesKey.Int(len(lh)))

		errG := errgroup.Group{}
		getTiles := func(ctx context.Context, tileIDs []storage.TileID, treeSize uint64) ([]*api.HashTile, error) {
			n, err := logStore.getTiles(ctx, tileIDs, treeSize)
			if err != nil {
				return nil, fmt.Errorf("getTiles: %w", err)
			}
			return n, nil
		}

		newSize, newRoot, tiles, err := storage.Integrate(ctx, getTiles, fromSeq, lh)
		if err != nil {
			return nil, fmt.Errorf("storage.Integrate: %v", err)
		}
		for k, v := range tiles {
			func(ctx context.Context, k storage.TileID, v *api.HashTile) {
				errG.Go(func() error {
					data, err := v.MarshalText()
					if err != nil {
						return err
					}
					return logStore.setTile(ctx, k.Level, k.Index, layout.PartialTileSize(k.Level, k.Index, newSize), data)
				})
			}(ctx, k, v)
		}
		if err := errG.Wait(); err != nil {
			return nil, err
		}
		slog.DebugContext(ctx, "New tree integrated", slog.Uint64("size", newSize), slog.String("root", fmt.Sprintf("%x", newRoot)))

		return newRoot, nil
	})
}

// updateEntryBundles adds the entries being integrated into the entry bundles.
//
// The right-most bundle will be grown, if it's partial, and/or new bundles will be created as required.
func (a *Appender) updateEntryBundles(ctx context.Context, fromSeq uint64, entries []storage.SequencedEntry) error {
	return otel.TraceErr(ctx, "tessera.storage.gcp.updateEntryBundles", tracer, func(ctx context.Context, span trace.Span) error {
		if len(entries) == 0 {
			return nil
		}

		numAdded := uint64(0)
		bundleIndex, entriesInBundle := fromSeq/layout.EntryBundleWidth, fromSeq%layout.EntryBundleWidth
		bundleWriter := &bytes.Buffer{}
		if entriesInBundle > 0 {
			// If the latest bundle is partial, we need to read the data it contains in for our newer, larger, bundle.
			part, err := a.logStore.getEntryBundle(ctx, uint64(bundleIndex), uint8(entriesInBundle))
			if err != nil {
				return err
			}

			if _, err := bundleWriter.Write(part); err != nil {
				return fmt.Errorf("bundleWriter: %v", err)
			}
		}

		seqErr := errgroup.Group{}

		// goSetEntryBundle is a function which uses seqErr to spin off a go-routine to write out an entry bundle.
		// It's used in the for loop below.
		goSetEntryBundle := func(ctx context.Context, bundleIndex uint64, p uint8, bundleRaw []byte) {
			seqErr.Go(func() error {
				if err := a.logStore.setEntryBundle(ctx, bundleIndex, p, bundleRaw); err != nil {
					return err
				}
				return nil
			})
		}

		// Add new entries to the bundle
		for _, e := range entries {
			if _, err := bundleWriter.Write(e.BundleData); err != nil {
				return fmt.Errorf("bundleWriter.Write: %v", err)
			}
			entriesInBundle++
			fromSeq++
			numAdded++
			if entriesInBundle == layout.EntryBundleWidth {
				//  This bundle is full, so we need to write it out...
				slog.DebugContext(ctx, "In-memory bundle is full, attempting write to GCS", slog.Uint64("bundleIndex", bundleIndex))
				goSetEntryBundle(ctx, bundleIndex, 0, bundleWriter.Bytes())
				// ... and prepare the next entry bundle for any remaining entries in the batch
				bundleIndex++
				entriesInBundle = 0
				// Don't use Reset/Truncate here - the backing []bytes is still being used by goSetEntryBundle above.
				bundleWriter = &bytes.Buffer{}
				slog.DebugContext(ctx, "Starting to fill in-memory bundle", slog.Uint64("bundleIndex", bundleIndex))
			}
		}
		// If we have a partial bundle remaining once we've added all the entries from the batch,
		// this needs writing out too.
		if entriesInBundle > 0 {
			slog.DebugContext(ctx, "Attempting to write in-memory partial bundle to GCS", slog.Uint64("bundleIndex", bundleIndex), slog.Uint64("entriesInBundle", entriesInBundle))
			goSetEntryBundle(ctx, bundleIndex, uint8(entriesInBundle), bundleWriter.Bytes())
		}
		return seqErr.Wait()
	})
}

// spannerCoordinator uses Cloud Spanner to provide
// a durable and thread/multi-process safe sequencer.
type spannerCoordinator struct {
	dbPool *spanner.Client
	// table returns the provided table name with the configured table prefix, if any, prepended.
	table          func(string) string
	maxOutstanding uint64

	// seqTableMaxBatchByteSize is the maximum byte size of a batch of entries to be written to the "V" field in the "Seq" table.
	seqTableMaxBatchByteSize int
}

// newSpannerCoordinator returns a new spannerSequencer struct which uses the provided
// spanner resource name for its spanner connection.
func newSpannerCoordinator(ctx context.Context, dbPool *spanner.Client, table func(string) string, maxOutstanding uint64) (*spannerCoordinator, error) {
	r := &spannerCoordinator{
		dbPool:                   dbPool,
		table:                    table,
		maxOutstanding:           maxOutstanding,
		seqTableMaxBatchByteSize: defaultSeqTableMaxBatchByteSize,
	}
	if err := r.checkDataCompatibility(ctx); err != nil {
		return nil, fmt.Errorf("schema is not compatible with this version of the Tessera library: %v", err)
	}
	return r, nil
}

// initDB ensures that the coordination DB is initialised correctly.
//
// The database schema consists of 5 tables:
//   - SeqCoord
//     This table only ever contains a single row which tracks the next available
//     sequence number.
//   - Seq
//     This table holds sequenced "batches" of entries. The batches are keyed
//     by the sequence number assigned to the first entry in the batch, and
//     each subsequent entry in the batch takes the numerically next sequence number.
//   - IntCoord
//     This table coordinates integration of the batches of entries stored in
//     Seq into the committed tree state.
//   - PubCoord
//     This table coordinates publication of checkpoints representing the
//     committed tree state.
//   - GCCoord
//     This table coordinates garbage collection of unneeded partial tiles
//     and entry bundles.
func initDB(ctx context.Context, spannerDB string, table func(string) string) error {
	return createAndPrepareTables(ctx, spannerDB,
		[]string{
			fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, compatibilityVersion INT64 NOT NULL) PRIMARY KEY (id)", table("Tessera")),
			fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, next INT64 NOT NULL,) PRIMARY KEY (id)", table("SeqCoord")),
			fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, seq INT64 NOT NULL, v BYTES(MAX),) PRIMARY KEY (id, seq)", table("Seq")),
			fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, seq INT64 NOT NULL, rootHash BYTES(32)) PRIMARY KEY (id)", table("IntCoord")),
			fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, publishedAt TIMESTAMP NOT NULL, size INT64) PRIMARY KEY (id)", table("PubCoord")),
			fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, fromSize INT64 NOT NULL) PRIMARY KEY (id)", table("GCCoord")),
		},
		[]string{
			fmt.Sprintf("ALTER TABLE %s ADD COLUMN IF NOT EXISTS size INT64", table("PubCoord")),
		},
		[][]*spanner.Mutation{
			{spanner.Insert(table("Tessera"), []string{"id", "compatibilityVersion"}, []any{0, SchemaCompatibilityVersion})},
			{spanner.Insert(table("SeqCoord"), []string{"id", "next"}, []any{0, 0})},
			{spanner.Insert(table("IntCoord"), []string{"id", "seq", "rootHash"}, []any{0, 0, rfc6962.DefaultHasher.EmptyRoot()})},
			{spanner.Insert(table("PubCoord"), []string{"id", "publishedAt", "size"}, []any{0, time.Unix(0, 0), 0})},
			{spanner.Insert(table("GCCoord"), []string{"id", "fromSize"}, []any{0, 0})},
		})
}

// checkDataCompatibility compares the Tessera library SchemaCompatibilityVersion with the one stored in the
// database, and returns an error if they are not identical.
func (s *spannerCoordinator) checkDataCompatibility(ctx context.Context) error {
	row, err := s.dbPool.Single().ReadRow(ctx, s.table("Tessera"), spanner.Key{0}, []string{"compatibilityVersion"})
	if err != nil {
		return fmt.Errorf("failed to read schema compatibilityVersion: %v", err)
	}
	var compat int64
	if err := row.Columns(&compat); err != nil {
		return fmt.Errorf("failed to scan schema compatibilityVersion: %v", err)
	}

	if compat != SchemaCompatibilityVersion {
		return fmt.Errorf("schema compatibilityVersion (%d) != library compatibilityVersion (%d)", compat, SchemaCompatibilityVersion)
	}
	return nil
}

// assignEntries durably assigns each of the passed-in entries an index in the log.
//
// Entries are allocated contiguous indices, in the order in which they appear in the entries parameter.
// This is achieved by storing the passed-in entries in the Seq table in Spanner, keyed by the
// index assigned to the first entry in the batch.
func (s *spannerCoordinator) assignEntries(ctx context.Context, entries []*tessera.Entry) error {
	return otel.TraceErr(ctx, "tessera.storage.gcp.assignEntries", tracer, func(ctx context.Context, span trace.Span) error {
		start := time.Now()
		defer func() {
			opsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("assignEntries")))
		}()

		span.SetAttributes(numEntriesKey.Int(len(entries)))

		span.AddEvent("Reading IntCoord:seq")
		// First grab the treeSize in a non-locking read-only fashion (we don't want to block/collide with integration).
		// We'll use this value to determine whether we need to apply back-pressure.
		var treeSize int64
		if row, err := s.dbPool.Single().ReadRowWithOptions(ctx, s.table("IntCoord"), spanner.Key{0}, []string{"seq"}, &spanner.ReadOptions{RequestTag: "tessera.op=assignEntries.treeSize"}); err != nil {
			return err
		} else {
			if err := row.Column(0, &treeSize); err != nil {
				return fmt.Errorf("failed to read integration coordination info: %v", err)
			}
		}
		span.SetAttributes(treeSizeKey.Int64(treeSize))

		var next int64 // Unfortunately, Spanner doesn't support uint64 so we'll have to cast around a bit.

		span.AddEvent("Starting ReadWriteTransaction")
		_, err := s.dbPool.ReadWriteTransactionWithOptions(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
			span.AddEvent("Reading SeqCoord:next")
			// First we need to grab the next available sequence number from the SeqCoord table.
			row, err := txn.ReadRowWithOptions(ctx, s.table("SeqCoord"), spanner.Key{0}, []string{"next"}, &spanner.ReadOptions{LockHint: spannerpb.ReadRequest_LOCK_HINT_EXCLUSIVE})
			if err != nil {
				return fmt.Errorf("failed to read SeqCoord: %w", err)
			}
			if err := row.Columns(&next); err != nil {
				return fmt.Errorf("failed to parse next column: %v", err)
			}

			// Check whether there are too many outstanding entries and we should apply
			// back-pressure.
			if outstanding := next - treeSize; outstanding > int64(s.maxOutstanding) {
				return tessera.ErrPushbackIntegration
			}

			span.AddEvent("Compiling mutations")
			var mutations []*spanner.Mutation
			next := uint64(next) // Shadow next with a uint64 version of the same value to save on casts.
			startFrom := next
			var sequencedEntries []storage.SequencedEntry
			currentBatchByteSize := 0
			// Assign provisional sequence numbers to entries.
			// We need to do this here in order to support serialisations which include the log position.
			for i, e := range entries {
				sequencedEntry := storage.SequencedEntry{
					BundleData: e.MarshalBundleData(startFrom + uint64(i)),
					LeafHash:   e.LeafHash(),
				}

				// If adding this entry would make the batch too big, we need to flush the original batch.
				if len(sequencedEntries) > 0 && currentBatchByteSize+len(sequencedEntry.BundleData) > s.seqTableMaxBatchByteSize {
					// Gob-encode the batch of entries and add it to the mutation.
					m, err := s.addSeqMutation(next, sequencedEntries)
					if err != nil {
						return fmt.Errorf("failed to addSeqMutation: %v", err)
					}
					mutations = append(mutations, m)
					next += uint64(len(sequencedEntries))

					// Reset our batch variables, and clear the batch slice now that it's been added to the
					// mutation.
					sequencedEntries = nil
					currentBatchByteSize = 0
				}

				sequencedEntries = append(sequencedEntries, sequencedEntry)
				currentBatchByteSize += len(sequencedEntry.BundleData)
			}

			// Insert the last batch of entries if there are any.
			if len(sequencedEntries) > 0 {
				m, err := s.addSeqMutation(next, sequencedEntries)
				if err != nil {
					return fmt.Errorf("failed to addSeqMutation: %v", err)
				}
				mutations = append(mutations, m)
				next += uint64(len(sequencedEntries))
			}

			// and update the next-available sequence number row in SeqCoord.
			mutations = append(mutations, spanner.Update(s.table("SeqCoord"), []string{"id", "next"}, []any{0, int64(next)}))

			span.AddEvent("Writing mutations")
			if err := txn.BufferWrite(mutations); err != nil {
				return fmt.Errorf("failed to apply TX: %v", err)
			}

			return nil
		}, spanner.TransactionOptions{TransactionTag: "tessera.op=assignEntries"})
		span.AddEvent("Finished ReadWriteTransaction")

		if err != nil {
			return fmt.Errorf("failed to flush batch: %w", err)
		}

		return nil
	}, trace.WithAttributes(otel.PeriodicKey.Bool(true)))
}

// addSeqMutation returns a mutation to the Seq table for the given sequence number and entries.
//
// The entries are gob-encoded and stored in the V column.
//
// The mutation is not written to the database; it is intended to be passed to a spanner.Transaction
// which will write it to the database.
func (s *spannerCoordinator) addSeqMutation(seq uint64, entries []storage.SequencedEntry) (*spanner.Mutation, error) {
	b := &bytes.Buffer{}
	if err := gob.NewEncoder(b).Encode(entries); err != nil {
		return nil, fmt.Errorf("failed to serialise batch: %v", err)
	}
	// Insert our newly sequenced batch of entries into Seq.
	return spanner.Insert(s.table("Seq"), []string{"id", "seq", "v"}, []any{0, int64(seq), b.Bytes()}), nil
}

// consumeEntries calls f with previously sequenced entries.
//
// Once f returns without error, the entries it was called with are considered to have been consumed and are
// removed from the Seq table.
//
// Returns true if some entries were consumed as a weak signal that there may be further entries waiting to be consumed.
func (s *spannerCoordinator) consumeEntries(ctx context.Context, limit uint64, f consumeFunc, forceUpdate bool) (bool, error) {
	if limit > math.MaxInt {
		limit = math.MaxInt
	}
	// Read this outside of the big write transaction below; we don't want to interfere with sequencing writes.
	seqLimit, err := s.nextIndex(ctx)
	if err != nil {
		return false, fmt.Errorf("nextIndex: %v", err)
	}
	return otel.Trace(ctx, "tessera.storage.gcp.consumeEntries", tracer, func(ctx context.Context, span trace.Span) (bool, error) {
		didWork := false
		_, err := s.dbPool.ReadWriteTransactionWithOptions(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
			span.AddEvent("Reading IntCoord:seq")
			// Figure out which is the starting index of sequenced entries to start consuming from.
			row, err := txn.ReadRowWithOptions(ctx, s.table("IntCoord"), spanner.Key{0}, []string{"seq"}, &spanner.ReadOptions{LockHint: spannerpb.ReadRequest_LOCK_HINT_EXCLUSIVE})
			if err != nil {
				return err
			}
			var fromSeq int64 // Spanner doesn't support uint64
			if err := row.Columns(&fromSeq); err != nil {
				return fmt.Errorf("failed to read integration coordination info: %v", err)
			}

			if fromSeq >= int64(seqLimit) {
				return nil
			}

			slog.DebugContext(ctx, "Consuming bundles", slog.Int64("fromSeq", fromSeq), slog.Uint64("toSeq", seqLimit-1))
			span.AddEvent("Reading entries from sequence table")
			// Now read the sequenced starting at the index we got above.
			rows := txn.Read(ctx, s.table("Seq"),
				spanner.KeyRange{Start: spanner.Key{0, fromSeq}, End: spanner.Key{0, int64(seqLimit)}, Kind: spanner.ClosedOpen},
				[]string{"seq", "v"})
			defer rows.Stop()

			seqsConsumed := []int64{}
			entries := make([]storage.SequencedEntry, 0, limit)
			orderCheck := fromSeq
			for {
				row, err := rows.Next()
				if row == nil || err == iterator.Done {
					break
				}

				var vGob []byte
				var seq int64 // spanner doesn't have uint64
				if err := row.Columns(&seq, &vGob); err != nil {
					return fmt.Errorf("failed to scan seq row: %v", err)
				}

				if orderCheck != seq {
					return fmt.Errorf("integrity fail - expected seq %d, but found %d", orderCheck, seq)
				}

				g := gob.NewDecoder(bytes.NewReader(vGob))
				b := []storage.SequencedEntry{}
				if err := g.Decode(&b); err != nil {
					return fmt.Errorf("failed to deserialise v: %v", err)
				}
				entries = append(entries, b...)
				seqsConsumed = append(seqsConsumed, seq)
				orderCheck += int64(len(b))
				if len(entries) >= int(limit) {
					break
				}
			}
			if len(seqsConsumed) == 0 && !forceUpdate {
				slog.DebugContext(ctx, "Found no rows to sequence")
				return nil
			}

			span.AddEvent(fmt.Sprintf("Integrating %d entries", len(entries)))
			// Call consumeFunc with the entries we've found
			newRoot, err := f(ctx, uint64(fromSeq), entries)
			if err != nil {
				return err
			}

			span.AddEvent("Creating mutations")
			// consumeFunc was successful, so we can update our coordination row, and delete the row(s) for
			// the then consumed entries.
			m := make([]*spanner.Mutation, 0)
			m = append(m, spanner.Update(s.table("IntCoord"), []string{"id", "seq", "rootHash"}, []any{0, int64(orderCheck), newRoot}))
			for _, c := range seqsConsumed {
				m = append(m, spanner.Delete(s.table("Seq"), spanner.Key{0, c}))
			}

			span.AddEvent("Writing mutations")
			if len(m) > 0 {
				if err := txn.BufferWrite(m); err != nil {
					return err
				}
			}

			didWork = true
			return nil
		}, spanner.TransactionOptions{TransactionTag: "tessera.op=consumeEntries"})
		if err != nil {
			return false, err
		}

		return didWork, nil
	})
}

// currentTree returns the size and root hash of the currently integrated tree.
func (s *spannerCoordinator) currentTree(ctx context.Context) (uint64, []byte, error) {
	row, err := s.dbPool.Single().ReadRowWithOptions(ctx, s.table("IntCoord"), spanner.Key{0}, []string{"seq", "rootHash"}, &spanner.ReadOptions{RequestTag: "tessera.op=currentTree"})
	if err != nil {
		return 0, nil, fmt.Errorf("failed to read IntCoord: %v", err)
	}
	var fromSeq int64 // Spanner doesn't support uint64
	var rootHash []byte
	if err := row.Columns(&fromSeq, &rootHash); err != nil {
		return 0, nil, fmt.Errorf("failed to read integration coordination info: %v", err)
	}

	return uint64(fromSeq), rootHash, nil
}

// nextIndex returns the next available index in the log.
func (s *spannerCoordinator) nextIndex(ctx context.Context) (uint64, error) {
	row, err := s.dbPool.Single().ReadRowWithOptions(ctx, s.table("SeqCoord"), spanner.Key{0}, []string{"next"}, &spanner.ReadOptions{RequestTag: "tessera.op=nextIndex"})
	if err != nil {
		return 0, fmt.Errorf("failed to read SeqCoord: %v", err)
	}
	var nextSeq int64 // Spanner doesn't support uint64
	if err := row.Columns(&nextSeq); err != nil {
		return 0, fmt.Errorf("failed to read sequence coordination info: %v", err)
	}

	return uint64(nextSeq), nil
}

// publishCheckpoint checks when the last checkpoint was published, and if appropriate, calls the provided
// function to publish a new one.
//
// Returns the time at which publishCheckpoint should be called again.
//
// A checkpoint will not be published if either:
//   - the currently published checkpoint was published less than minStaleActive ago
//   - the new checkpoint is the same size as the currently published one, AND the currently published checkpoint
//     was published less than minStaleRepub ago.
//
// This function uses PubCoord with an exclusive lock to guarantee that only one tessera instance can attempt to publish
// a checkpoint at any given time.
func (s *spannerCoordinator) publishCheckpoint(ctx context.Context, minStaleActive, minStaleRepub time.Duration, f func(context.Context, uint64, []byte) error) (time.Time, error) {
	return otel.Trace(ctx, "tessera.storage.gcp.publishCheckpoint", tracer, func(ctx context.Context, span trace.Span) (time.Time, error) {
		// outcomeAttr records the outcome of the checkpoint publication attempt for metrics and traces.
		var outcomeAttr attribute.KeyValue
		var nextPubAt time.Time
		start := time.Now()

		currentSize, rootHash, err := s.currentTree(ctx)
		if err != nil {
			return time.Time{}, fmt.Errorf("failed to get current tree: %v", err)
		}

		span.AddEvent("Starting ReadWriteTransaction")
		if _, err := s.dbPool.ReadWriteTransactionWithOptions(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
			// Reset outcome attributes from any prior transaction attempts.
			outcomeAttr = outcomeTypeKey.String("success")

			span.AddEvent("Reading PubCoord")
			pRow, err := txn.ReadRowWithOptions(ctx, s.table("PubCoord"), spanner.Key{0}, []string{"publishedAt", "size"}, &spanner.ReadOptions{LockHint: spannerpb.ReadRequest_LOCK_HINT_EXCLUSIVE})
			if err != nil {
				return fmt.Errorf("failed to read PubCoord: %w", err)
			}
			var lastSize spanner.NullInt64
			var pubAt time.Time
			if err := pRow.Columns(&pubAt, &lastSize); err != nil {
				return fmt.Errorf("failed to parse PubCoord: %v", err)
			}

			if lastSize.Valid && int64(currentSize) < lastSize.Int64 {
				// Our view is stale, abort.
				outcomeAttr = outcomeTypeKey.String("aborted")
				return nil
			}

			cpAge := time.Since(pubAt)
			if cpAge < minStaleActive {
				slog.DebugContext(ctx, "publishCheckpoint: last checkpoint published too recently, not publishing new checkpoint", slog.Duration("cpAge", cpAge), slog.Duration("minStaleActive", minStaleActive))
				outcomeAttr = outcomeTypeKey.String("skipped")
				nextPubAt = pubAt.Add(minStaleActive)
				return nil
			}

			shouldPublish := minStaleRepub > 0 && cpAge >= minStaleRepub
			if !shouldPublish {
				if !lastSize.Valid {
					// If we don't know the last published size, we should probably publish to be safe/self-heal.
					shouldPublish = true
				} else if currentSize > uint64(lastSize.Int64) {
					shouldPublish = true
				}
			}

			if !shouldPublish {
				slog.DebugContext(ctx, "publishCheckpoint: skipping publish because tree hasn't grown and previous checkpoint is too recent")
				outcomeAttr = outcomeTypeKey.String("skipped_no_growth")
				nextPubAt = start.Add(minStaleActive)
				if minStaleRepub > 0 && nextPubAt.After(pubAt.Add(minStaleRepub)) {
					nextPubAt = pubAt.Add(minStaleRepub)
				}
				return nil
			}

			slog.DebugContext(ctx, "publishCheckpoint: updating checkpoint", slog.Duration("cpAge", cpAge))

			span.AddEvent("Publishing checkpoint")
			if err := f(ctx, currentSize, rootHash); err != nil {
				return err
			}
			span.AddEvent("Updating PubCoord")
			pubAt = time.Now()
			nextPubAt = pubAt.Add(minStaleActive)
			if err := txn.BufferWrite([]*spanner.Mutation{spanner.Update(s.table("PubCoord"), []string{"id", "publishedAt", "size"}, []any{0, pubAt, int64(currentSize)})}); err != nil {
				return err
			}

			return nil
		}, spanner.TransactionOptions{TransactionTag: "tessera.op=publishCheckpoint"}); err != nil {
			span.SetAttributes(outcomeTypeKey.String("error"))
			publishCount.Add(ctx, 1, metric.WithAttributes(outcomeTypeKey.String("error")))
			return time.Time{}, err
		}
		opsHistogram.Record(ctx, time.Since(start).Milliseconds(), metric.WithAttributes(opNameKey.String("publishCheckpoint")))
		span.SetAttributes(outcomeAttr)
		publishCount.Add(ctx, 1, metric.WithAttributes(outcomeAttr))
		return nextPubAt, nil
	})
}

// garbageCollect will identify up to maxBundles unneeded partial entry bundles (and any unneeded partial tiles which sit above them in the tree) and
// call the provided function to remove them.
//
// Uses the `GCCoord` table to ensure that only one binary is actively garbage collecting at any given time, and to track progress so that we don't
// needlessly attempt to GC over regions which have already been cleaned.
func (s *spannerCoordinator) garbageCollect(ctx context.Context, treeSize uint64, maxBundles uint, deleteWithPrefix func(ctx context.Context, prefix string) error, entriesPath func(uint64, uint8) string) error {
	// number of concurrent GCS deletes to perform while collecting garbage.
	const numWorkers = 1

	_, err := s.dbPool.ReadWriteTransactionWithOptions(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
		row, err := txn.ReadRowWithOptions(ctx, s.table("GCCoord"), spanner.Key{0}, []string{"fromSize"}, &spanner.ReadOptions{LockHint: spannerpb.ReadRequest_LOCK_HINT_EXCLUSIVE})
		if err != nil {
			return fmt.Errorf("failed to read GCCoord: %w", err)
		}
		var fs int64
		if err := row.Columns(&fs); err != nil {
			return fmt.Errorf("failed to parse row contents: %v", err)
		}
		fromSize := uint64(fs)

		if fromSize == treeSize {
			return nil
		}

		d := uint(0)
		work := make(chan string, maxBundles*3)

		eg := errgroup.Group{}
		eg.Go(func() error {
			// GC the tree in "vertical" chunks defined by entry bundles.
			for ri := range layout.Range(fromSize, treeSize-fromSize, treeSize) {
				// Only known-full bundles are in-scope for GC, so exit if the current bundle is partial or
				// we've reached our limit of chunks.
				if ri.Partial > 0 || d > maxBundles {
					break
				}

				// GC any partial versions of the entry bundle itself and the tile which sits immediately above it.
				work <- entriesPath(ri.Index, 0) + ".p/"
				work <- layout.TilePath(0, ri.Index, 0) + ".p/"

				fromSize += uint64(ri.N)
				d++

				// Now consider (only) the part of the tree which sits above the bundle.
				// We'll walk up the parent tiles for as a long as we're tracing the right-hand
				// edge of a perfect subtree.
				// This gives the property we'll only visit each parent tile once, rather than up to 256 times.
				pL, pIdx := uint64(0), ri.Index
				for isLastLeafInParent(pIdx) {
					// Move our coordinates up to the parent
					pL, pIdx = pL+1, pIdx>>layout.TileHeight
					// GC any partial versions of the parent tile.
					work <- layout.TilePath(pL, pIdx, 0) + ".p/"
				}
			}
			close(work)
			return nil
		})

		for range numWorkers {
			eg.Go(func() error {
				errs := []error{}
				for prefix := range work {
					if err := deleteWithPrefix(ctx, prefix); err != nil {
						errs = append(errs, err)
					}
				}
				return errors.Join(errs...)
			})
		}
		if err := eg.Wait(); err != nil {
			return fmt.Errorf("failed to delete one or more objects: %v", err)
		}

		if err := txn.BufferWrite([]*spanner.Mutation{spanner.Update(s.table("GCCoord"), []string{"id", "fromSize"}, []any{0, int64(fromSize)})}); err != nil {
			return err
		}

		return nil
	}, spanner.TransactionOptions{TransactionTag: "tessera.op=garbageCollect"})
	return err
}

// isLastLeafInParent returns true if a tile with the provided index is the final child node of a
// (hypothetical) full parent tile.
func isLastLeafInParent(i uint64) bool {
	return i%layout.TileWidth == layout.TileWidth-1
}

// gcsStorage knows how to store and retrieve objects from GCS.
type gcsStorage struct {
	bucket       string
	bucketPrefix string
	gcsClient    *gcs.Client
}

// getObject returns the data and generation of the specified object, or an error.
func (s *gcsStorage) getObject(ctx context.Context, obj string) ([]byte, *gcs.ReaderObjectAttrs, error) {
	return otel.Trace2(ctx, "tessera.storage.gcp.getObject", tracer, func(ctx context.Context, span trace.Span) ([]byte, *gcs.ReaderObjectAttrs, error) {
		if s.bucketPrefix != "" {
			obj = filepath.Join(s.bucketPrefix, obj)
		}

		span.SetAttributes(objectPathKey.String(obj))

		r, err := s.gcsClient.Bucket(s.bucket).Object(obj).NewReader(ctx)
		if err != nil {
			return nil, nil, fmt.Errorf("getObject: failed to create reader for object %q in bucket %q: %w", obj, s.bucket, err)
		}

		d, err := io.ReadAll(r)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to read %q: %v", obj, err)
		}
		return d, &r.Attrs, r.Close()
	})
}

// setObject stores the provided data in the specified object, optionally gated by a condition.
//
// cond can be used to specify preconditions for the write (e.g. write iff not exists, write iff
// current generation is X, etc.), or nil can be passed if no preconditions are desired.
//
// Note that when preconditions are specified and are not met, an error will be returned *unless*
// the currently stored data is bit-for-bit identical to the data to-be-written.
// This is intended to provide idempotentency for writes.
func (s *gcsStorage) setObject(ctx context.Context, objName string, data []byte, cond *gcs.Conditions, contType string, cacheCtl string) error {
	return otel.TraceErr(ctx, "tessera.storage.gcp.setObject", tracer, func(ctx context.Context, span trace.Span) error {
		if s.bucketPrefix != "" {
			objName = filepath.Join(s.bucketPrefix, objName)
		}

		span.SetAttributes(objectPathKey.String(objName))

		bkt := s.gcsClient.Bucket(s.bucket)
		obj := bkt.Object(objName)

		var w *gcs.Writer
		if cond == nil {
			w = obj.NewWriter(ctx)

		} else {
			w = obj.If(*cond).NewWriter(ctx)
		}
		w.ContentType = contType
		w.CacheControl = cacheCtl
		// Limit the amount of memory used for buffers, see https://pkg.go.dev/cloud.google.com/go/storage#Writer
		w.ChunkSize = len(data) + 1024
		if _, err := w.Write(data); err != nil {
			return fmt.Errorf("failed to write object %q to bucket %q: %w", objName, s.bucket, err)
		}

		if err := w.Close(); err != nil {
			// If we run into a precondition failure error, check that the object
			// which exists contains the same content that we want to write.
			// If so, we can consider this write to be idempotently successful.
			preconditionFailed := false

			// Helpfully, the mechanism for detecting a failed precodition differs depending
			// on whether you're using the HTTP or gRPC GCS client, so test both.
			if ee, ok := err.(*googleapi.Error); ok && ee.Code == http.StatusPreconditionFailed {
				preconditionFailed = true
			} else if st, ok := status.FromError(err); ok && st.Code() == codes.FailedPrecondition {
				preconditionFailed = true
			}
			if preconditionFailed {
				existing, existingAttr, err := s.getObject(ctx, objName)
				if err != nil {
					return fmt.Errorf("failed to fetch existing content for %q: %v", objName, err)
				}
				if !bytes.Equal(existing, data) {
					span.AddEvent("Non-idempotent write")
					slog.ErrorContext(ctx, "Resource non-idempotent write", slog.String("objName", objName), slog.String("diff", cmp.Diff(existing, data)))
					return fmt.Errorf("precondition failed: resource content for %q (@%d) differs from data to-be-written", objName, existingAttr.Generation)
				}

				span.AddEvent("Idempotent write")
				slog.DebugContext(ctx, "setObject: identical resource already exists", slog.String("objName", objName))
				return nil
			}

			return fmt.Errorf("failed to close write on %q: %v", objName, err)
		}
		return nil
	})
}

// deleteObjectsWithPrefix removes any objects with the provided prefix from GCS.
func (s *gcsStorage) deleteObjectsWithPrefix(ctx context.Context, objPrefix string) error {
	return otel.TraceErr(ctx, "tessera.storage.gcp.deleteObject", tracer, func(ctx context.Context, span trace.Span) error {
		if s.bucketPrefix != "" {
			objPrefix = filepath.Join(s.bucketPrefix, objPrefix)
		}
		span.SetAttributes(objectPathKey.String(objPrefix))

		bkt := s.gcsClient.Bucket(s.bucket)

		errs := []error(nil)
		q := &gcs.Query{
			Prefix:     objPrefix,
			Projection: gcs.ProjectionNoACL,
		}
		if err := q.SetAttrSelection([]string{"Name"}); err != nil {
			return fmt.Errorf("failed to set attr selection: %v", err)
		}
		it := bkt.Objects(ctx, q)
		for {
			attr, err := it.Next()
			if err != nil {
				if err == iterator.Done {
					break
				}
				return err
			}
			slog.DebugContext(ctx, "Deleting object", slog.String("name", attr.Name))
			if err := bkt.Object(attr.Name).Delete(ctx); err != nil {
				errs = append(errs, err)
			}
		}

		return errors.Join(errs...)
	})
}

// MigrationWriter creates a new GCP storage for the MigrationTarget lifecycle mode.
func (s *Storage) MigrationWriter(ctx context.Context, opts *tessera.MigrationOptions) (migrate.MigrationWriter, tessera.LogReader, error) {
	var err error
	if s.cfg.GCSClient == nil {
		s.cfg.GCSClient, err = gcs.NewClient(ctx, gcs.WithJSONReads())
		if err != nil {
			return nil, nil, fmt.Errorf("failed to create GCS client: %v", err)
		}
	}

	if s.cfg.SpannerClient == nil {
		s.cfg.SpannerClient, err = spanner.NewClient(ctx, s.cfg.Spanner)
		if err != nil {
			return nil, nil, fmt.Errorf("failed to connect to Spanner: %v", err)
		}
	}
	table := func(t string) string {
		return s.cfg.SpannerTablePrefix + t
	}
	if err := initDB(ctx, s.cfg.Spanner, table); err != nil {
		return nil, nil, fmt.Errorf("failed to verify/init Spanner schema: %v", err)
	}

	seq, err := newSpannerCoordinator(ctx, s.cfg.SpannerClient, table, 0)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create Spanner sequencer: %v", err)
	}
	m := &MigrationStorage{
		s:            s,
		dbPool:       seq.dbPool,
		table:        table,
		bundleHasher: opts.LeafHasher(),
		sequencer:    seq,
		logStore: &logResourceStore{
			objStore: &gcsStorage{
				gcsClient:    s.cfg.GCSClient,
				bucket:       s.cfg.Bucket,
				bucketPrefix: s.cfg.BucketPrefix,
			},
			entriesPath: opts.EntriesPath(),
		},
	}

	r := &LogReader{
		lrs: *m.logStore,
		integratedSize: func(ctx context.Context) (uint64, error) {
			s, _, err := m.sequencer.currentTree(ctx)
			return s, err
		},
		nextIndex: func(context.Context) (uint64, error) {
			return 0, nil
		},
	}
	return m, r, nil
}

// MigrationStorgage implements the tessera.MigrationTarget lifecycle contract.
type MigrationStorage struct {
	s      *Storage
	dbPool *spanner.Client
	// table returns the provided table name with the configured table prefix, if any, prepended.
	table        func(string) string
	bundleHasher func([]byte) ([][]byte, error)
	sequencer    sequencer
	logStore     *logResourceStore
}

var _ migrate.MigrationWriter = &MigrationStorage{}

func (m *MigrationStorage) AwaitIntegration(ctx context.Context, sourceSize uint64) ([]byte, error) {
	t := time.NewTicker(time.Second)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-t.C:
			from, _, err := m.sequencer.currentTree(ctx)
			if err != nil && !errors.Is(err, os.ErrNotExist) {
				slog.WarnContext(ctx, "readTreeState failed", slog.Any("error", err))
				continue
			}
			slog.InfoContext(ctx, "Integrate", slog.Uint64("from", from), slog.Uint64("sourceSize", sourceSize))
			newSize, newRoot, err := m.buildTree(ctx, sourceSize)
			if err != nil {
				slog.WarnContext(ctx, "integrate failed", slog.Any("error", err))
			}
			if newSize == sourceSize {
				slog.InfoContext(ctx, "Integrated", slog.Uint64("newSize", newSize), slog.String("newRoot", fmt.Sprintf("%x", newRoot)))
				return newRoot, nil
			}
		}
	}
}

func (m *MigrationStorage) SetEntryBundle(ctx context.Context, index uint64, partial uint8, bundle []byte) error {
	return m.logStore.setEntryBundle(ctx, index, partial, bundle)
}

func (m *MigrationStorage) IntegratedSize(ctx context.Context) (uint64, error) {
	sz, _, err := m.sequencer.currentTree(ctx)
	return sz, err
}

func (m *MigrationStorage) fetchLeafHashes(ctx context.Context, from, to, sourceSize uint64) ([][]byte, error) {
	// TODO(al): Make this configurable.
	const maxBundles = 100

	toBeAdded := sync.Map{}
	eg := errgroup.Group{}
	n := 0
	for ri := range layout.Range(from, to, sourceSize) {
		eg.Go(func() error {
			b, err := m.logStore.getEntryBundle(ctx, ri.Index, ri.Partial)
			if err != nil {
				return fmt.Errorf("getEntryBundle(%d.%d): %v", ri.Index, ri.Partial, err)
			}

			bh, err := m.bundleHasher(b)
			if err != nil {
				return fmt.Errorf("bundleHasherFunc for bundle index %d: %v", ri.Index, err)
			}
			if l := len(bh); l < int(ri.First+ri.N) {
				return fmt.Errorf("bundle index %d has fewer entries than expected (%d < %d)", ri.Index, l, ri.First+ri.N)
			}
			toBeAdded.Store(ri.Index, bh[ri.First:ri.First+ri.N])
			return nil
		})
		n++
		if n >= maxBundles {
			break
		}
	}
	if err := eg.Wait(); err != nil {
		return nil, err
	}

	lh := make([][]byte, 0, maxBundles)
	for i := from / layout.EntryBundleWidth; ; i++ {
		v, ok := toBeAdded.LoadAndDelete(i)
		if !ok {
			break
		}
		bh := v.([][]byte)
		lh = append(lh, bh...)
	}

	return lh, nil
}

func (m *MigrationStorage) buildTree(ctx context.Context, sourceSize uint64) (uint64, []byte, error) {
	var newSize uint64
	var newRoot []byte

	_, err := m.dbPool.ReadWriteTransaction(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
		// Figure out which is the starting index of sequenced entries to start consuming from.
		row, err := txn.ReadRowWithOptions(ctx, m.table("IntCoord"), spanner.Key{0}, []string{"seq", "rootHash"}, &spanner.ReadOptions{LockHint: spannerpb.ReadRequest_LOCK_HINT_EXCLUSIVE})
		if err != nil {
			return err
		}
		var fromSeq int64 // Spanner doesn't support uint64
		var rootHash []byte
		if err := row.Columns(&fromSeq, &rootHash); err != nil {
			return fmt.Errorf("failed to read integration coordination info: %v", err)
		}

		from := uint64(fromSeq)
		slog.DebugContext(ctx, "Integrating", slog.Uint64("from", from))
		lh, err := m.fetchLeafHashes(ctx, from, sourceSize, sourceSize)
		if err != nil {
			return fmt.Errorf("fetchLeafHashes(%d, %d, %d): %v", from, sourceSize, sourceSize, err)
		}

		if len(lh) == 0 {
			slog.InfoContext(ctx, "Integrate: nothing to do, nothing done")
			// Set these to the current state of the tree so we reflect that in buildTree's return values.
			newSize, newRoot = from, rootHash
			return nil
		}

		added := uint64(len(lh))
		slog.InfoContext(ctx, "Integrate: adding entries to existing tree", slog.Int("count", len(lh)), slog.Uint64("from", from))
		newRoot, err = integrate(ctx, from, lh, m.logStore)
		if err != nil {
			slog.WarnContext(ctx, "integrate failed", slog.Any("error", err))
			return fmt.Errorf("integrate failed: %v", err)
		}
		newSize = from + added
		slog.InfoContext(ctx, "Integrate: added entries", slog.Uint64("added", added))

		// integration was successful, so we can update our coordination row
		muts := []*spanner.Mutation{spanner.Update(m.table("IntCoord"), []string{"id", "seq", "rootHash"}, []any{0, int64(from + added), newRoot})}
		return txn.BufferWrite(muts)
	})

	if err != nil {
		return 0, nil, err
	}
	return newSize, newRoot, nil
}

// createAndPrepareTables applies the passed in list of DDL statements and groups of mutations.
//
// This is intended to be used to create and initialise Spanner instances on first use.
// DDL should likely be of the form "CREATE TABLE IF NOT EXISTS".
// Mutation groups should likey be one or more spanner.Insert operations - AlreadyExists errors will be silently ignored.
func createAndPrepareTables(ctx context.Context, spannerDB string, ddl []string, alter []string, mutations [][]*spanner.Mutation) error {
	adminClient, err := database.NewDatabaseAdminClient(ctx)
	if err != nil {
		return err
	}
	defer func() {
		if err := adminClient.Close(); err != nil {
			slog.WarnContext(ctx, "adminClient.Close() failed", slog.Any("error", err))
		}
	}()

	op, err := adminClient.UpdateDatabaseDdl(ctx, &adminpb.UpdateDatabaseDdlRequest{
		Database:   spannerDB,
		Statements: ddl,
	})
	if err != nil {
		return fmt.Errorf("failed to create tables: %v", err)
	}
	if err := op.Wait(ctx); err != nil {
		return err
	}

	if len(alter) > 0 {
		// The spannertest emulator appears to ignore IF NOT EXISTS in ALTER DATABASE statements, so
		// we'll apply each update individually and ignore any AlreadyExists errors we see.
		for _, a := range alter {
			op, err := adminClient.UpdateDatabaseDdl(ctx, &adminpb.UpdateDatabaseDdlRequest{
				Database:   spannerDB,
				Statements: []string{a},
			})
			if err != nil { // && spanner.ErrCode(err) != codes.AlreadyExists {
				return fmt.Errorf("updateDatabaseDdl: %v", err)
			}
			if err := op.Wait(ctx); err != nil && spanner.ErrCode(err) != codes.AlreadyExists {
				return fmt.Errorf("failed to alter table: %v", err)
			}
		}
	}

	dbPool, err := spanner.NewClient(ctx, spannerDB)
	if err != nil {
		return fmt.Errorf("failed to connect to Spanner: %v", err)
	}
	defer dbPool.Close()

	// Set default values for a newly initialised schema using passed in mutation groups.
	// Note that this will only succeed if no row exists, so there's no danger of "resetting" an existing log.
	for _, mg := range mutations {
		if _, err := dbPool.Apply(ctx, mg); err != nil && spanner.ErrCode(err) != codes.AlreadyExists {
			return err
		}
	}
	return nil
}