From 5775fd47683b854063c968e1943566e844e97c2e Mon Sep 17 00:00:00 2001 From: Erin Perrine Date: Fri, 21 Aug 2026 16:01:28 +0000 Subject: [PATCH 1/2] storage/gcp: skip schema DDL when the Spanner schema already exists initDB (used by Appender and MigrationWriter) and antispam.NewAntispam currently apply their CREATE TABLE IF NOT EXISTS / ALTER TABLE ... ADD COLUMN IF NOT EXISTS statements through a DatabaseAdmin client, and then insert the seed rows, every time they are called - not only when a log is first created. Spanner executes DDL as schema-update operations, which take on the order of seconds each and are serialised per database even when IF NOT EXISTS means they end up changing nothing, so simply restarting a log costs several schema-update operations. When many logs share one database via SpannerTablePrefix this adds up quickly, and concurrent restarts queue up behind one another. Make initDB first do a cheap read-only check using the data client which Appender/MigrationWriter have already created: if every table it would create exists, PubCoord already has its size column, every seed row is present, and the stored compatibilityVersion matches the library's, then the DDL and seeding would be a no-op and are skipped. If anything is missing, different, or cannot be read, initDB carries on and creates or migrates the schema exactly as before, so the check can only ever elide work which would have done nothing. The same client is now also used to apply the seed mutations, rather than a temporary one. The antispam storage gets the same treatment for its FollowCoord and IDSeq tables. NewAntispam now resolves its Spanner client (the provided one, or a new one) before initialising the schema, so that one client serves the check, the seeding, and the returned storage. There is no API or schema change, and behaviour against an empty or partially initialised database is unchanged. Tests cover the check against initialised, empty, and several partially initialised or mismatched databases, and show that re-running initDB and NewAntispam against an existing schema succeeds and leaves existing state alone. The spannertest emulator rejects CREATE TABLE for an existing table regardless of IF NOT EXISTS, so those re-runs would fail if any DDL were still being applied. --- storage/gcp/antispam/gcp.go | 74 +++++++++---- storage/gcp/antispam/gcp_test.go | 146 +++++++++++++++++++++++- storage/gcp/gcp.go | 68 ++++++++++-- storage/gcp/gcp_test.go | 184 +++++++++++++++++++++++++++++-- 4 files changed, 425 insertions(+), 47 deletions(-) diff --git a/storage/gcp/antispam/gcp.go b/storage/gcp/antispam/gcp.go index 207561fa2..bdf02c614 100644 --- a/storage/gcp/antispam/gcp.go +++ b/storage/gcp/antispam/gcp.go @@ -121,19 +121,6 @@ func NewAntispam(ctx context.Context, spannerDB string, opts AntispamOpts) (*Ant return opts.SpannerTablePrefix + t } - if err := createAndPrepareTables( - ctx, spannerDB, opts.SpannerClient, - []string{ - fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, nextIdx INT64 NOT NULL) PRIMARY KEY (id)", table("FollowCoord")), - fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (h BYTES(32) NOT NULL, idx INT64 NOT NULL) PRIMARY KEY (h)", table("IDSeq")), - }, - [][]*spanner.Mutation{ - {spanner.Insert(table("FollowCoord"), []string{"id", "nextIdx"}, []any{0, 0})}, - }, - ); err != nil { - return nil, fmt.Errorf("failed to create tables: %v", err) - } - db := opts.SpannerClient if db == nil { var err error @@ -143,6 +130,13 @@ func NewAntispam(ctx context.Context, spannerDB string, opts AntispamOpts) (*Ant } } + if err := initDB(ctx, spannerDB, db, table); err != nil { + if opts.SpannerClient == nil { + db.Close() + } + return nil, fmt.Errorf("failed to create tables: %v", err) + } + r := &AntispamStorage{ opts: opts, dbPool: db, @@ -487,13 +481,55 @@ func (f *follower) EntriesProcessed(ctx context.Context) (uint64, error) { return uint64(nextIdx), nil } +// initDB ensures that the antispam DB is initialised correctly. +// +// Spanner executes DDL statements as schema-update operations, which are slow (and +// serialised per database) even when IF NOT EXISTS means they end up changing nothing, +// so the DDL and seeding below are skipped entirely if a cheap read via dbPool shows +// that the schema they would create is already fully present - see schemaInitialised. +func initDB(ctx context.Context, spannerDB string, dbPool *spanner.Client, table func(string) string) error { + if schemaInitialised(ctx, dbPool, table) { + return nil + } + // Note that schemaInitialised needs to be kept in sync with any changes to the statements or mutations below. + return createAndPrepareTables( + ctx, spannerDB, dbPool, + []string{ + fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, nextIdx INT64 NOT NULL) PRIMARY KEY (id)", table("FollowCoord")), + fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (h BYTES(32) NOT NULL, idx INT64 NOT NULL) PRIMARY KEY (h)", table("IDSeq")), + }, + [][]*spanner.Mutation{ + {spanner.Insert(table("FollowCoord"), []string{"id", "nextIdx"}, []any{0, 0})}, + }, + ) +} + +// schemaInitialised returns true if the schema which initDB creates is already fully present in +// the database: both tables exist and the FollowCoord seed row is present. +// +// If anything is missing or simply cannot be read, it returns false so that initDB goes on to +// create the schema exactly as it would have done anyway. I.e. this check can only ever cause +// initDB to skip work which would have been a no-op, so it must be kept in sync with the DDL +// and mutations in initDB. +func schemaInitialised(ctx context.Context, dbPool *spanner.Client, table func(string) string) bool { + // A successful read of the seed row shows that the table, the named columns, and the row + // itself are all present. + if _, err := dbPool.Single().ReadRow(ctx, table("FollowCoord"), spanner.Key{0}, []string{"id", "nextIdx"}); err != nil { + return false + } + // IDSeq has no seed row, so just check that the table itself exists. + if err := dbPool.Single().ReadWithOptions(ctx, table("IDSeq"), spanner.AllKeys(), []string{"h", "idx"}, &spanner.ReadOptions{Limit: 1}).Do(func(*spanner.Row) error { return nil }); err != nil { + return false + } + return true +} + // 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. -// If dbPool is non-nil it is used to apply the mutations (and is not closed); otherwise a -// temporary client is created for the duration of this call. +// dbPool is used to apply the mutations, and is not closed. func createAndPrepareTables(ctx context.Context, spannerDB string, dbPool *spanner.Client, ddl []string, mutations [][]*spanner.Mutation) error { adminClient, err := database.NewDatabaseAdminClient(ctx) if err != nil { @@ -516,14 +552,6 @@ func createAndPrepareTables(ctx context.Context, spannerDB string, dbPool *spann return err } - if dbPool == nil { - 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 { diff --git a/storage/gcp/antispam/gcp_test.go b/storage/gcp/antispam/gcp_test.go index 391d2ca1f..0a6d7786c 100644 --- a/storage/gcp/antispam/gcp_test.go +++ b/storage/gcp/antispam/gcp_test.go @@ -25,6 +25,8 @@ import ( "log/slog" "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/spannertest" "github.com/transparency-dev/tessera" "github.com/transparency-dev/tessera/api" @@ -105,9 +107,8 @@ func TestAntispamStorage(t *testing.T) { t.Run(test.name, func(t *testing.T) { closeDB := newSpannerDB(t) defer closeDB() - const spannerDB = "projects/p/instances/i/databases/d" if test.sharedClient { - c, err := spanner.NewClient(t.Context(), spannerDB) + c, err := spanner.NewClient(t.Context(), testSpannerDB) if err != nil { t.Fatalf("spanner.NewClient: %v", err) } @@ -117,7 +118,7 @@ func TestAntispamStorage(t *testing.T) { t.Cleanup(c.Close) test.opts.SpannerClient = c } - as, err := NewAntispam(t.Context(), spannerDB, test.opts) + as, err := NewAntispam(t.Context(), testSpannerDB, test.opts) if err != nil { t.Fatalf("NewAntispam: %v", err) } @@ -194,7 +195,7 @@ func TestAntispamSharedClientWrongDatabase(t *testing.T) { } defer c.Close() - if _, err := NewAntispam(t.Context(), "projects/p/instances/i/databases/d", AntispamOpts{SpannerClient: c}); err == nil { + if _, err := NewAntispam(t.Context(), testSpannerDB, AntispamOpts{SpannerClient: c}); err == nil { t.Error("NewAntispam accepted a SpannerClient connected to a different database, want error") } } @@ -220,7 +221,7 @@ func TestAntispamPushbackRecovers(t *testing.T) { t.Run(test.name, func(t *testing.T) { closeDB := newSpannerDB(t) defer closeDB() - as, err := NewAntispam(t.Context(), "projects/p/instances/i/databases/d", test.opts) + as, err := NewAntispam(t.Context(), testSpannerDB, test.opts) if err != nil { t.Fatalf("NewAntispam: %v", err) } @@ -283,6 +284,141 @@ func TestAntispamPushbackRecovers(t *testing.T) { } } +func TestNewAntispamExistingSchema(t *testing.T) { + ctx := t.Context() + closeDB := newSpannerDB(t) + defer closeDB() + + db, err := spanner.NewClient(ctx, testSpannerDB) + if err != nil { + t.Fatalf("spanner.NewClient: %v", err) + } + defer db.Close() + opts := AntispamOpts{SpannerTablePrefix: "Tenant1_", SpannerClient: db} + + if schemaInitialised(ctx, db, prefixTable(opts.SpannerTablePrefix)) { + t.Fatal("schemaInitialised: got true on empty DB, want false") + } + if _, err := NewAntispam(ctx, testSpannerDB, opts); err != nil { + t.Fatalf("NewAntispam on empty DB: %v", err) + } + if !schemaInitialised(ctx, db, prefixTable(opts.SpannerTablePrefix)) { + t.Fatal("schemaInitialised: got false after NewAntispam, want true") + } + if _, err := db.Apply(ctx, []*spanner.Mutation{spanner.Update(opts.SpannerTablePrefix+"FollowCoord", []string{"id", "nextIdx"}, []any{0, 42})}); err != nil { + t.Fatalf("Apply: %v", err) + } + + // The spannertest emulator does not honour IF NOT EXISTS on CREATE TABLE statements, so the + // second NewAntispam below would fail if it attempted to apply any DDL - check that this is + // still the case, so that this test can't pass vacuously if the emulator changes. + if err := createAndPrepareTables(ctx, testSpannerDB, db, []string{"CREATE TABLE IF NOT EXISTS Tenant1_IDSeq (h BYTES(32) NOT NULL, idx INT64 NOT NULL) PRIMARY KEY (h)"}, nil); err == nil { + t.Skip("spannertest now honours CREATE TABLE IF NOT EXISTS, so this test can no longer tell whether NewAntispam applied DDL") + } + + // Opening antispam storage which has already been initialised, e.g. when restarting a log, should + // succeed without applying any DDL (see above), and must not disturb existing state. + as, err := NewAntispam(ctx, testSpannerDB, opts) + if err != nil { + t.Fatalf("NewAntispam on existing schema: %v", err) + } + f := as.Follower(testBundleHasher) + if got, err := f.EntriesProcessed(ctx); err != nil || got != 42 { + t.Fatalf("EntriesProcessed: got %d, %v, want 42, nil", got, err) + } +} + +func TestSchemaInitialised(t *testing.T) { + for _, test := range []struct { + name string + // prep, if set, is used to modify a database in which NewAntispam has already created the (unprefixed) schema. + prep func(ctx context.Context, t *testing.T, db *spanner.Client) + // table identifies the tables to check, defaults to unprefixed. + table func(string) string + want bool + }{ + { + name: "initialised", + want: true, + }, { + name: "not initialised: no tables with this prefix", + table: prefixTable("Other_"), + want: false, + }, { + name: "missing seed row", + prep: func(ctx context.Context, t *testing.T, db *spanner.Client) { + if _, err := db.Apply(ctx, []*spanner.Mutation{spanner.Delete("FollowCoord", spanner.Key{0})}); err != nil { + t.Fatalf("Apply: %v", err) + } + }, + want: false, + }, { + name: "missing unseeded table", + prep: func(ctx context.Context, t *testing.T, db *spanner.Client) { + applyDDL(t, "DROP TABLE IDSeq") + }, + want: false, + }, + } { + t.Run(test.name, func(t *testing.T) { + ctx := t.Context() + closeDB := newSpannerDB(t) + defer closeDB() + db, err := spanner.NewClient(ctx, testSpannerDB) + if err != nil { + t.Fatalf("spanner.NewClient: %v", err) + } + defer db.Close() + if _, err := NewAntispam(ctx, testSpannerDB, AntispamOpts{SpannerClient: db}); err != nil { + t.Fatalf("NewAntispam: %v", err) + } + if test.prep != nil { + test.prep(ctx, t, db) + } + if test.table == nil { + test.table = prefixTable("") + } + if got := schemaInitialised(ctx, db, test.table); got != test.want { + t.Fatalf("schemaInitialised: got %t, want %t", got, test.want) + } + }) + } +} + +// testSpannerDB is the resource name of the database served by the spannertest emulator in these tests. +const testSpannerDB = "projects/p/instances/i/databases/d" + +// prefixTable returns a function which returns the provided table name with prefix prepended. +func prefixTable(prefix string) func(string) string { + return func(table string) string { + return prefix + table + } +} + +// applyDDL applies the provided DDL statements directly to testSpannerDB. +func applyDDL(t *testing.T, statements ...string) { + t.Helper() + adminClient, err := database.NewDatabaseAdminClient(t.Context()) + if err != nil { + t.Fatalf("NewDatabaseAdminClient: %v", err) + } + defer func() { + if err := adminClient.Close(); err != nil { + t.Logf("adminClient.Close: %v", err) + } + }() + op, err := adminClient.UpdateDatabaseDdl(t.Context(), &adminpb.UpdateDatabaseDdlRequest{ + Database: testSpannerDB, + Statements: statements, + }) + if err != nil { + t.Fatalf("UpdateDatabaseDdl(%q): %v", statements, err) + } + if err := op.Wait(t.Context()); err != nil { + t.Fatalf("UpdateDatabaseDdl(%q): %v", statements, err) + } +} + func newSpannerDB(t *testing.T) func() { t.Helper() srv, err := spannertest.NewServer("localhost:0") diff --git a/storage/gcp/gcp.go b/storage/gcp/gcp.go index 7725470cf..5b4007494 100644 --- a/storage/gcp/gcp.go +++ b/storage/gcp/gcp.go @@ -255,7 +255,7 @@ func (s *Storage) Appender(ctx context.Context, opts *tessera.AppendOptions) (*t table := func(t string) string { return s.cfg.SpannerTablePrefix + t } - if err := initDB(ctx, s.cfg.Spanner, table); err != nil { + if err := initDB(ctx, s.cfg.Spanner, s.cfg.SpannerClient, table); err != nil { return nil, nil, fmt.Errorf("failed to verify/init Spanner schema: %v", err) } @@ -820,8 +820,19 @@ func newSpannerCoordinator(ctx context.Context, dbPool *spanner.Client, table fu // - 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, +// +// Spanner executes DDL statements as schema-update operations, which are slow (and +// serialised per database) even when IF NOT EXISTS means they end up changing nothing, +// so the DDL and seeding below are skipped entirely if a cheap read via dbPool shows +// that the schema they would create is already fully present - see schemaInitialised. +// This keeps re-opening an existing log fast, which matters particularly when many +// logs share one database via SpannerTablePrefix. +func initDB(ctx context.Context, spannerDB string, dbPool *spanner.Client, table func(string) string) error { + if schemaInitialised(ctx, dbPool, table) { + return nil + } + // Note that schemaInitialised needs to be kept in sync with any changes to the statements or mutations below. + return createAndPrepareTables(ctx, spannerDB, dbPool, []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")), @@ -842,6 +853,46 @@ func initDB(ctx context.Context, spannerDB string, table func(string) string) er }) } +// schemaInitialised returns true if the schema which initDB creates is already fully present in +// the database: every table exists, PubCoord has its size column, every seed row is present, +// and the stored compatibilityVersion matches this version of the library (so that any schema +// migration which a future initDB performs for older versions will still be run). +// +// If anything is missing, different, or simply cannot be read, it returns false so that initDB +// goes on to create/migrate the schema exactly as it would have done anyway. I.e. this check +// can only ever cause initDB to skip work which would have been a no-op, so it must be kept in +// sync with the DDL and mutations in initDB. +func schemaInitialised(ctx context.Context, dbPool *spanner.Client, table func(string) string) bool { + row, err := dbPool.Single().ReadRow(ctx, table("Tessera"), spanner.Key{0}, []string{"compatibilityVersion"}) + if err != nil { + return false + } + var compat int64 + if err := row.Columns(&compat); err != nil || compat != SchemaCompatibilityVersion { + return false + } + // A successful read of each seed row shows that the table, the named columns, and the row + // itself are all present. + for _, seed := range []struct { + table string + cols []string + }{ + {table: "SeqCoord", cols: []string{"id", "next"}}, + {table: "IntCoord", cols: []string{"id", "seq", "rootHash"}}, + {table: "PubCoord", cols: []string{"id", "publishedAt", "size"}}, + {table: "GCCoord", cols: []string{"id", "fromSize"}}, + } { + if _, err := dbPool.Single().ReadRow(ctx, table(seed.table), spanner.Key{0}, seed.cols); err != nil { + return false + } + } + // Seq has no seed row, so just check that the table itself exists. + if err := dbPool.Single().ReadWithOptions(ctx, table("Seq"), spanner.AllKeys(), []string{"id", "seq"}, &spanner.ReadOptions{Limit: 1}).Do(func(*spanner.Row) error { return nil }); err != nil { + return false + } + return true +} + // 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 { @@ -1470,7 +1521,7 @@ func (s *Storage) MigrationWriter(ctx context.Context, opts *tessera.MigrationOp table := func(t string) string { return s.cfg.SpannerTablePrefix + t } - if err := initDB(ctx, s.cfg.Spanner, table); err != nil { + if err := initDB(ctx, s.cfg.Spanner, s.cfg.SpannerClient, table); err != nil { return nil, nil, fmt.Errorf("failed to verify/init Spanner schema: %v", err) } @@ -1657,7 +1708,8 @@ func (m *MigrationStorage) buildTree(ctx context.Context, sourceSize uint64) (ui // 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 { +// dbPool is used to apply the mutations, and is not closed. +func createAndPrepareTables(ctx context.Context, spannerDB string, dbPool *spanner.Client, ddl []string, alter []string, mutations [][]*spanner.Mutation) error { adminClient, err := database.NewDatabaseAdminClient(ctx) if err != nil { return err @@ -1696,12 +1748,6 @@ func createAndPrepareTables(ctx context.Context, spannerDB string, ddl []string, } } - 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 { diff --git a/storage/gcp/gcp_test.go b/storage/gcp/gcp_test.go index 33540a268..e6d0eebc6 100644 --- a/storage/gcp/gcp_test.go +++ b/storage/gcp/gcp_test.go @@ -30,6 +30,8 @@ import ( "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/spannertest" gcs "cloud.google.com/go/storage" "github.com/google/go-cmp/cmp" @@ -46,6 +48,9 @@ func init() { slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))) } +// testSpannerDB is the resource name of the database served by the spannertest emulator in these tests. +const testSpannerDB = "projects/p/instances/i/databases/d" + func newSpannerDB(t *testing.T) (*spanner.Client, func()) { t.Helper() return newSpannerDBWithPrefix(t, "") @@ -58,7 +63,9 @@ func prefixTable(prefix string) func(string) string { } } -func newSpannerDBWithPrefix(t *testing.T, tablePrefix string) (*spanner.Client, func()) { +// newEmptySpannerDB starts a new spannertest emulator with no schema, points SPANNER_EMULATOR_HOST +// at it, and returns a client connected to testSpannerDB along with a func to shut the emulator down. +func newEmptySpannerDB(t *testing.T) (*spanner.Client, func()) { t.Helper() srv, err := spannertest.NewServer("localhost:0") if err != nil { @@ -68,17 +75,44 @@ func newSpannerDBWithPrefix(t *testing.T, tablePrefix string) (*spanner.Client, t.Fatalf("Setenv: %v", err) } - id := "projects/p/instances/i/databases/d" - if err := initDB(t.Context(), id, prefixTable(tablePrefix)); err != nil { - t.Fatalf("initDB: %v", err) - } - - c, err := spanner.NewClient(t.Context(), id) + c, err := spanner.NewClient(t.Context(), testSpannerDB) if err != nil { t.Fatalf("NewClient: %v", err) } return c, srv.Close +} + +func newSpannerDBWithPrefix(t *testing.T, tablePrefix string) (*spanner.Client, func()) { + t.Helper() + c, close := newEmptySpannerDB(t) + if err := initDB(t.Context(), testSpannerDB, c, prefixTable(tablePrefix)); err != nil { + t.Fatalf("initDB: %v", err) + } + return c, close +} +// applyDDL applies the provided DDL statements directly to testSpannerDB. +func applyDDL(t *testing.T, statements ...string) { + t.Helper() + adminClient, err := database.NewDatabaseAdminClient(t.Context()) + if err != nil { + t.Fatalf("NewDatabaseAdminClient: %v", err) + } + defer func() { + if err := adminClient.Close(); err != nil { + t.Logf("adminClient.Close: %v", err) + } + }() + op, err := adminClient.UpdateDatabaseDdl(t.Context(), &adminpb.UpdateDatabaseDdlRequest{ + Database: testSpannerDB, + Statements: statements, + }) + if err != nil { + t.Fatalf("UpdateDatabaseDdl(%q): %v", statements, err) + } + if err := op.Wait(t.Context()); err != nil { + t.Fatalf("UpdateDatabaseDdl(%q): %v", statements, err) + } } func TestSpannerSequencerAssignEntries(t *testing.T) { @@ -165,7 +199,7 @@ func TestSpannerTablePrefixValidation(t *testing.T) { t.Run(test.name, func(t *testing.T) { _, err := New(ctx, Config{ Bucket: "bucket", - Spanner: "projects/p/instances/i/databases/d", + Spanner: testSpannerDB, SpannerTablePrefix: test.prefix, }) if gotErr := err != nil; gotErr != test.wantErr { @@ -325,6 +359,140 @@ func TestCheckDataCompatibility(t *testing.T) { } } +func TestSchemaInitialised(t *testing.T) { + for _, test := range []struct { + name string + // prep, if set, is used to modify a database in which initDB has already created the (unprefixed) schema. + prep func(ctx context.Context, t *testing.T, db *spanner.Client) + // table identifies the tables to check, defaults to unprefixed. + table func(string) string + want bool + }{ + { + name: "initialised", + want: true, + }, { + name: "initialised: NULL PubCoord.size as left by the ADD COLUMN migration", + prep: func(ctx context.Context, t *testing.T, db *spanner.Client) { + if _, err := db.Apply(ctx, []*spanner.Mutation{spanner.Update("PubCoord", []string{"id", "size"}, []any{0, spanner.NullInt64{}})}); err != nil { + t.Fatalf("Apply: %v", err) + } + }, + want: true, + }, { + name: "not initialised: no tables with this prefix", + table: prefixTable("Other_"), + want: false, + }, { + name: "missing compatibilityVersion row", + prep: func(ctx context.Context, t *testing.T, db *spanner.Client) { + if _, err := db.Apply(ctx, []*spanner.Mutation{spanner.Delete("Tessera", spanner.Key{0})}); err != nil { + t.Fatalf("Apply: %v", err) + } + }, + want: false, + }, { + name: "different compatibilityVersion", + prep: func(ctx context.Context, t *testing.T, db *spanner.Client) { + if _, err := db.Apply(ctx, []*spanner.Mutation{spanner.Update("Tessera", []string{"id", "compatibilityVersion"}, []any{0, SchemaCompatibilityVersion + 1})}); err != nil { + t.Fatalf("Apply: %v", err) + } + }, + want: false, + }, { + name: "missing seed row", + prep: func(ctx context.Context, t *testing.T, db *spanner.Client) { + if _, err := db.Apply(ctx, []*spanner.Mutation{spanner.Delete("GCCoord", spanner.Key{0})}); err != nil { + t.Fatalf("Apply: %v", err) + } + }, + want: false, + }, { + name: "missing PubCoord.size column: older schema in need of migration", + prep: func(ctx context.Context, t *testing.T, db *spanner.Client) { + applyDDL(t, "ALTER TABLE PubCoord DROP COLUMN size") + }, + want: false, + }, { + name: "missing unseeded table", + prep: func(ctx context.Context, t *testing.T, db *spanner.Client) { + applyDDL(t, "DROP TABLE Seq") + }, + want: false, + }, + } { + t.Run(test.name, func(t *testing.T) { + ctx := t.Context() + db, close := newSpannerDB(t) + defer close() + if test.prep != nil { + test.prep(ctx, t, db) + } + if test.table == nil { + test.table = prefixTable("") + } + if got := schemaInitialised(ctx, db, test.table); got != test.want { + t.Fatalf("schemaInitialised: got %t, want %t", got, test.want) + } + }) + } +} + +func TestInitDBExistingSchema(t *testing.T) { + ctx := t.Context() + // newEmptySpannerDB rather than newSpannerDB so that the initial initDB below is explicit. + db, close := newEmptySpannerDB(t) + defer close() + + if schemaInitialised(ctx, db, prefixTable("")) { + t.Fatal("schemaInitialised: got true on empty DB, want false") + } + if err := initDB(ctx, testSpannerDB, db, prefixTable("")); err != nil { + t.Fatalf("initDB on empty DB: %v", err) + } + if !schemaInitialised(ctx, db, prefixTable("")) { + t.Fatal("schemaInitialised: got false after initDB, want true") + } + + // The spannertest emulator does not honour IF NOT EXISTS on CREATE TABLE statements, so the + // further calls to initDB below would fail if they attempted to apply any DDL - check that this + // is still the case, so that this test can't pass vacuously if the emulator changes. + if err := createAndPrepareTables(ctx, testSpannerDB, db, []string{"CREATE TABLE IF NOT EXISTS Tessera (id INT64 NOT NULL, compatibilityVersion INT64 NOT NULL) PRIMARY KEY (id)"}, nil, nil); err == nil { + t.Skip("spannertest now honours CREATE TABLE IF NOT EXISTS, so this test can no longer tell whether initDB applied DDL") + } + + seq, err := newSpannerCoordinator(ctx, db, prefixTable(""), 1000) + if err != nil { + t.Fatalf("newSpannerCoordinator: %v", err) + } + entries := []*tessera.Entry{} + for i := range 5 { + entries = append(entries, tessera.NewEntry(fmt.Appendf(nil, "item %d", i))) + } + if err := seq.assignEntries(ctx, entries); err != nil { + t.Fatalf("assignEntries: %v", err) + } + + // Re-initialising an existing schema, e.g. when restarting a log, should succeed without + // applying any DDL (see above), and must not disturb existing state. + for i := range 2 { + if err := initDB(ctx, testSpannerDB, db, prefixTable("")); err != nil { + t.Fatalf("initDB on existing schema (attempt %d): %v", i, err) + } + } + seq2, err := newSpannerCoordinator(ctx, db, prefixTable(""), 1000) + if err != nil { + t.Fatalf("newSpannerCoordinator after re-init: %v", err) + } + next, err := seq2.nextIndex(ctx) + if err != nil { + t.Fatalf("nextIndex: %v", err) + } + if want := uint64(len(entries)); next != want { + t.Errorf("nextIndex after re-init: got %d, want %d", next, want) + } +} + func makeTile(t *testing.T, size uint64) *api.HashTile { t.Helper() r := &api.HashTile{Nodes: make([][]byte, size)} From 7072d2b804e0a30158bfe0a614ced904ed4c2102 Mon Sep 17 00:00:00 2001 From: Erin Perrine Date: Fri, 21 Aug 2026 17:20:30 +0000 Subject: [PATCH 2/2] Inline the antispam schema check, leave createAndPrepareTables unchanged, shorten comments --- storage/gcp/antispam/gcp.go | 67 ++++++++++++++------------------ storage/gcp/antispam/gcp_test.go | 18 ++++----- storage/gcp/gcp.go | 36 +++++++---------- storage/gcp/gcp_test.go | 22 +++++------ 4 files changed, 59 insertions(+), 84 deletions(-) diff --git a/storage/gcp/antispam/gcp.go b/storage/gcp/antispam/gcp.go index bdf02c614..60643804d 100644 --- a/storage/gcp/antispam/gcp.go +++ b/storage/gcp/antispam/gcp.go @@ -130,11 +130,23 @@ func NewAntispam(ctx context.Context, spannerDB string, opts AntispamOpts) (*Ant } } - if err := initDB(ctx, spannerDB, db, table); err != nil { - if opts.SpannerClient == nil { - db.Close() + // Skip the (slow, even when no-op) DDL if the schema is already present. Keep schemaInitialised in sync with this. + if !schemaInitialised(ctx, db, table) { + if err := createAndPrepareTables( + ctx, spannerDB, db, + []string{ + fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, nextIdx INT64 NOT NULL) PRIMARY KEY (id)", table("FollowCoord")), + fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (h BYTES(32) NOT NULL, idx INT64 NOT NULL) PRIMARY KEY (h)", table("IDSeq")), + }, + [][]*spanner.Mutation{ + {spanner.Insert(table("FollowCoord"), []string{"id", "nextIdx"}, []any{0, 0})}, + }, + ); err != nil { + if opts.SpannerClient == nil { + db.Close() + } + return nil, fmt.Errorf("failed to create tables: %v", err) } - return nil, fmt.Errorf("failed to create tables: %v", err) } r := &AntispamStorage{ @@ -481,43 +493,13 @@ func (f *follower) EntriesProcessed(ctx context.Context) (uint64, error) { return uint64(nextIdx), nil } -// initDB ensures that the antispam DB is initialised correctly. -// -// Spanner executes DDL statements as schema-update operations, which are slow (and -// serialised per database) even when IF NOT EXISTS means they end up changing nothing, -// so the DDL and seeding below are skipped entirely if a cheap read via dbPool shows -// that the schema they would create is already fully present - see schemaInitialised. -func initDB(ctx context.Context, spannerDB string, dbPool *spanner.Client, table func(string) string) error { - if schemaInitialised(ctx, dbPool, table) { - return nil - } - // Note that schemaInitialised needs to be kept in sync with any changes to the statements or mutations below. - return createAndPrepareTables( - ctx, spannerDB, dbPool, - []string{ - fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (id INT64 NOT NULL, nextIdx INT64 NOT NULL) PRIMARY KEY (id)", table("FollowCoord")), - fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (h BYTES(32) NOT NULL, idx INT64 NOT NULL) PRIMARY KEY (h)", table("IDSeq")), - }, - [][]*spanner.Mutation{ - {spanner.Insert(table("FollowCoord"), []string{"id", "nextIdx"}, []any{0, 0})}, - }, - ) -} - -// schemaInitialised returns true if the schema which initDB creates is already fully present in -// the database: both tables exist and the FollowCoord seed row is present. -// -// If anything is missing or simply cannot be read, it returns false so that initDB goes on to -// create the schema exactly as it would have done anyway. I.e. this check can only ever cause -// initDB to skip work which would have been a no-op, so it must be kept in sync with the DDL -// and mutations in initDB. +// schemaInitialised reports whether the tables and seed row NewAntispam creates are all present. +// Any error reads as false, so NewAntispam falls back to creating them. func schemaInitialised(ctx context.Context, dbPool *spanner.Client, table func(string) string) bool { - // A successful read of the seed row shows that the table, the named columns, and the row - // itself are all present. if _, err := dbPool.Single().ReadRow(ctx, table("FollowCoord"), spanner.Key{0}, []string{"id", "nextIdx"}); err != nil { return false } - // IDSeq has no seed row, so just check that the table itself exists. + // IDSeq has no seed row; just check the table exists. if err := dbPool.Single().ReadWithOptions(ctx, table("IDSeq"), spanner.AllKeys(), []string{"h", "idx"}, &spanner.ReadOptions{Limit: 1}).Do(func(*spanner.Row) error { return nil }); err != nil { return false } @@ -529,7 +511,8 @@ func schemaInitialised(ctx context.Context, dbPool *spanner.Client, table func(s // 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. -// dbPool is used to apply the mutations, and is not closed. +// If dbPool is non-nil it is used to apply the mutations (and is not closed); otherwise a +// temporary client is created for the duration of this call. func createAndPrepareTables(ctx context.Context, spannerDB string, dbPool *spanner.Client, ddl []string, mutations [][]*spanner.Mutation) error { adminClient, err := database.NewDatabaseAdminClient(ctx) if err != nil { @@ -552,6 +535,14 @@ func createAndPrepareTables(ctx context.Context, spannerDB string, dbPool *spann return err } + if dbPool == nil { + 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 { diff --git a/storage/gcp/antispam/gcp_test.go b/storage/gcp/antispam/gcp_test.go index 0a6d7786c..b2f5a6a7e 100644 --- a/storage/gcp/antispam/gcp_test.go +++ b/storage/gcp/antispam/gcp_test.go @@ -309,15 +309,13 @@ func TestNewAntispamExistingSchema(t *testing.T) { t.Fatalf("Apply: %v", err) } - // The spannertest emulator does not honour IF NOT EXISTS on CREATE TABLE statements, so the - // second NewAntispam below would fail if it attempted to apply any DDL - check that this is - // still the case, so that this test can't pass vacuously if the emulator changes. + // spannertest rejects CREATE TABLE IF NOT EXISTS on an existing table, so a re-open that ran DDL + // would fail below; guard against the emulator changing and this passing vacuously. if err := createAndPrepareTables(ctx, testSpannerDB, db, []string{"CREATE TABLE IF NOT EXISTS Tenant1_IDSeq (h BYTES(32) NOT NULL, idx INT64 NOT NULL) PRIMARY KEY (h)"}, nil); err == nil { t.Skip("spannertest now honours CREATE TABLE IF NOT EXISTS, so this test can no longer tell whether NewAntispam applied DDL") } - // Opening antispam storage which has already been initialised, e.g. when restarting a log, should - // succeed without applying any DDL (see above), and must not disturb existing state. + // Re-opening must apply no DDL (see above) and leave existing state alone. as, err := NewAntispam(ctx, testSpannerDB, opts) if err != nil { t.Fatalf("NewAntispam on existing schema: %v", err) @@ -331,9 +329,9 @@ func TestNewAntispamExistingSchema(t *testing.T) { func TestSchemaInitialised(t *testing.T) { for _, test := range []struct { name string - // prep, if set, is used to modify a database in which NewAntispam has already created the (unprefixed) schema. + // prep modifies a DB in which NewAntispam has created the unprefixed schema. prep func(ctx context.Context, t *testing.T, db *spanner.Client) - // table identifies the tables to check, defaults to unprefixed. + // table defaults to unprefixed. table func(string) string want bool }{ @@ -385,17 +383,17 @@ func TestSchemaInitialised(t *testing.T) { } } -// testSpannerDB is the resource name of the database served by the spannertest emulator in these tests. +// testSpannerDB is the database served by the spannertest emulator. const testSpannerDB = "projects/p/instances/i/databases/d" -// prefixTable returns a function which returns the provided table name with prefix prepended. +// prefixTable returns a func which prepends prefix to a table name. func prefixTable(prefix string) func(string) string { return func(table string) string { return prefix + table } } -// applyDDL applies the provided DDL statements directly to testSpannerDB. +// applyDDL applies DDL directly to testSpannerDB. func applyDDL(t *testing.T, statements ...string) { t.Helper() adminClient, err := database.NewDatabaseAdminClient(t.Context()) diff --git a/storage/gcp/gcp.go b/storage/gcp/gcp.go index 5b4007494..64976eb1b 100644 --- a/storage/gcp/gcp.go +++ b/storage/gcp/gcp.go @@ -820,19 +820,12 @@ func newSpannerCoordinator(ctx context.Context, dbPool *spanner.Client, table fu // - GCCoord // This table coordinates garbage collection of unneeded partial tiles // and entry bundles. -// -// Spanner executes DDL statements as schema-update operations, which are slow (and -// serialised per database) even when IF NOT EXISTS means they end up changing nothing, -// so the DDL and seeding below are skipped entirely if a cheap read via dbPool shows -// that the schema they would create is already fully present - see schemaInitialised. -// This keeps re-opening an existing log fast, which matters particularly when many -// logs share one database via SpannerTablePrefix. func initDB(ctx context.Context, spannerDB string, dbPool *spanner.Client, table func(string) string) error { + // Skip the (slow, even when no-op) DDL if the schema is already present. Keep schemaInitialised in sync with this. if schemaInitialised(ctx, dbPool, table) { return nil } - // Note that schemaInitialised needs to be kept in sync with any changes to the statements or mutations below. - return createAndPrepareTables(ctx, spannerDB, dbPool, + 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")), @@ -853,15 +846,8 @@ func initDB(ctx context.Context, spannerDB string, dbPool *spanner.Client, table }) } -// schemaInitialised returns true if the schema which initDB creates is already fully present in -// the database: every table exists, PubCoord has its size column, every seed row is present, -// and the stored compatibilityVersion matches this version of the library (so that any schema -// migration which a future initDB performs for older versions will still be run). -// -// If anything is missing, different, or simply cannot be read, it returns false so that initDB -// goes on to create/migrate the schema exactly as it would have done anyway. I.e. this check -// can only ever cause initDB to skip work which would have been a no-op, so it must be kept in -// sync with the DDL and mutations in initDB. +// schemaInitialised reports whether everything initDB creates is present (tables, PubCoord.size, +// seed rows) at this SchemaCompatibilityVersion. Any error reads as false, so initDB runs as before. func schemaInitialised(ctx context.Context, dbPool *spanner.Client, table func(string) string) bool { row, err := dbPool.Single().ReadRow(ctx, table("Tessera"), spanner.Key{0}, []string{"compatibilityVersion"}) if err != nil { @@ -871,8 +857,7 @@ func schemaInitialised(ctx context.Context, dbPool *spanner.Client, table func(s if err := row.Columns(&compat); err != nil || compat != SchemaCompatibilityVersion { return false } - // A successful read of each seed row shows that the table, the named columns, and the row - // itself are all present. + // Reading each seed row proves the table, columns, and row exist. for _, seed := range []struct { table string cols []string @@ -886,7 +871,7 @@ func schemaInitialised(ctx context.Context, dbPool *spanner.Client, table func(s return false } } - // Seq has no seed row, so just check that the table itself exists. + // Seq has no seed row; just check the table exists. if err := dbPool.Single().ReadWithOptions(ctx, table("Seq"), spanner.AllKeys(), []string{"id", "seq"}, &spanner.ReadOptions{Limit: 1}).Do(func(*spanner.Row) error { return nil }); err != nil { return false } @@ -1708,8 +1693,7 @@ func (m *MigrationStorage) buildTree(ctx context.Context, sourceSize uint64) (ui // 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. -// dbPool is used to apply the mutations, and is not closed. -func createAndPrepareTables(ctx context.Context, spannerDB string, dbPool *spanner.Client, ddl []string, alter []string, mutations [][]*spanner.Mutation) error { +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 @@ -1748,6 +1732,12 @@ func createAndPrepareTables(ctx context.Context, spannerDB string, dbPool *spann } } + 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 { diff --git a/storage/gcp/gcp_test.go b/storage/gcp/gcp_test.go index e6d0eebc6..902a7d21c 100644 --- a/storage/gcp/gcp_test.go +++ b/storage/gcp/gcp_test.go @@ -48,7 +48,7 @@ func init() { slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))) } -// testSpannerDB is the resource name of the database served by the spannertest emulator in these tests. +// testSpannerDB is the database served by the spannertest emulator. const testSpannerDB = "projects/p/instances/i/databases/d" func newSpannerDB(t *testing.T) (*spanner.Client, func()) { @@ -63,8 +63,7 @@ func prefixTable(prefix string) func(string) string { } } -// newEmptySpannerDB starts a new spannertest emulator with no schema, points SPANNER_EMULATOR_HOST -// at it, and returns a client connected to testSpannerDB along with a func to shut the emulator down. +// newEmptySpannerDB starts a schemaless spannertest emulator and returns a client and a shutdown func. func newEmptySpannerDB(t *testing.T) (*spanner.Client, func()) { t.Helper() srv, err := spannertest.NewServer("localhost:0") @@ -91,7 +90,7 @@ func newSpannerDBWithPrefix(t *testing.T, tablePrefix string) (*spanner.Client, return c, close } -// applyDDL applies the provided DDL statements directly to testSpannerDB. +// applyDDL applies DDL directly to testSpannerDB. func applyDDL(t *testing.T, statements ...string) { t.Helper() adminClient, err := database.NewDatabaseAdminClient(t.Context()) @@ -362,9 +361,9 @@ func TestCheckDataCompatibility(t *testing.T) { func TestSchemaInitialised(t *testing.T) { for _, test := range []struct { name string - // prep, if set, is used to modify a database in which initDB has already created the (unprefixed) schema. + // prep modifies a DB in which initDB has created the unprefixed schema. prep func(ctx context.Context, t *testing.T, db *spanner.Client) - // table identifies the tables to check, defaults to unprefixed. + // table defaults to unprefixed. table func(string) string want bool }{ @@ -440,7 +439,6 @@ func TestSchemaInitialised(t *testing.T) { func TestInitDBExistingSchema(t *testing.T) { ctx := t.Context() - // newEmptySpannerDB rather than newSpannerDB so that the initial initDB below is explicit. db, close := newEmptySpannerDB(t) defer close() @@ -454,10 +452,9 @@ func TestInitDBExistingSchema(t *testing.T) { t.Fatal("schemaInitialised: got false after initDB, want true") } - // The spannertest emulator does not honour IF NOT EXISTS on CREATE TABLE statements, so the - // further calls to initDB below would fail if they attempted to apply any DDL - check that this - // is still the case, so that this test can't pass vacuously if the emulator changes. - if err := createAndPrepareTables(ctx, testSpannerDB, db, []string{"CREATE TABLE IF NOT EXISTS Tessera (id INT64 NOT NULL, compatibilityVersion INT64 NOT NULL) PRIMARY KEY (id)"}, nil, nil); err == nil { + // spannertest rejects CREATE TABLE IF NOT EXISTS on an existing table, so a re-open that ran DDL + // would fail below; guard against the emulator changing and this passing vacuously. + if err := createAndPrepareTables(ctx, testSpannerDB, []string{"CREATE TABLE IF NOT EXISTS Tessera (id INT64 NOT NULL, compatibilityVersion INT64 NOT NULL) PRIMARY KEY (id)"}, nil, nil); err == nil { t.Skip("spannertest now honours CREATE TABLE IF NOT EXISTS, so this test can no longer tell whether initDB applied DDL") } @@ -473,8 +470,7 @@ func TestInitDBExistingSchema(t *testing.T) { t.Fatalf("assignEntries: %v", err) } - // Re-initialising an existing schema, e.g. when restarting a log, should succeed without - // applying any DDL (see above), and must not disturb existing state. + // Re-running initDB must apply no DDL (see above) and leave existing state alone. for i := range 2 { if err := initDB(ctx, testSpannerDB, db, prefixTable("")); err != nil { t.Fatalf("initDB on existing schema (attempt %d): %v", i, err)