diff --git a/storage/gcp/antispam/gcp.go b/storage/gcp/antispam/gcp.go index 207561fa2..60643804d 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,25 @@ func NewAntispam(ctx context.Context, spannerDB string, opts AntispamOpts) (*Ant } } + // 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) + } + } + r := &AntispamStorage{ opts: opts, dbPool: db, @@ -487,6 +493,19 @@ func (f *follower) EntriesProcessed(ctx context.Context) (uint64, error) { return uint64(nextIdx), nil } +// 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 { + if _, err := dbPool.Single().ReadRow(ctx, table("FollowCoord"), spanner.Key{0}, []string{"id", "nextIdx"}); err != nil { + return false + } + // 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 + } + 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. diff --git a/storage/gcp/antispam/gcp_test.go b/storage/gcp/antispam/gcp_test.go index 391d2ca1f..b2f5a6a7e 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,139 @@ 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) + } + + // 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") + } + + // 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) + } + 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 modifies a DB in which NewAntispam has created the unprefixed schema. + prep func(ctx context.Context, t *testing.T, db *spanner.Client) + // table 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 database served by the spannertest emulator. +const testSpannerDB = "projects/p/instances/i/databases/d" + +// 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 DDL 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..64976eb1b 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,7 +820,11 @@ 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 { +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 + } 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")), @@ -842,6 +846,38 @@ func initDB(ctx context.Context, spannerDB string, table func(string) string) er }) } +// 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 { + return false + } + var compat int64 + if err := row.Columns(&compat); err != nil || compat != SchemaCompatibilityVersion { + return false + } + // Reading each seed row proves the table, columns, and row exist. + 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; 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 + } + 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 +1506,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) } diff --git a/storage/gcp/gcp_test.go b/storage/gcp/gcp_test.go index 33540a268..902a7d21c 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 database served by the spannertest emulator. +const testSpannerDB = "projects/p/instances/i/databases/d" + func newSpannerDB(t *testing.T) (*spanner.Client, func()) { t.Helper() return newSpannerDBWithPrefix(t, "") @@ -58,7 +63,8 @@ func prefixTable(prefix string) func(string) string { } } -func newSpannerDBWithPrefix(t *testing.T, tablePrefix string) (*spanner.Client, func()) { +// 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") if err != nil { @@ -68,17 +74,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 DDL 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 +198,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 +358,137 @@ func TestCheckDataCompatibility(t *testing.T) { } } +func TestSchemaInitialised(t *testing.T) { + for _, test := range []struct { + name string + // prep modifies a DB in which initDB has created the unprefixed schema. + prep func(ctx context.Context, t *testing.T, db *spanner.Client) + // table 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() + 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") + } + + // 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") + } + + 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-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) + } + } + 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)}