storage/gcp: skip schema DDL when the Spanner schema already exists

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.

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")

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)
 	}