storage/gcp/antispam: add optional SpannerClient to AntispamOpts

Allows callers to provide an existing Spanner client to NewAntispam
instead of it always creating its own, mirroring Config.SpannerClient
on the GCP log storage. This lets multi-tenant deployments share a
single client (and its session pool) across the log storage and any
number of antispam instances in the same database.

An injected client must be connected to the database named by the
spannerDB parameter; this is validated at construction since a
mismatched client would otherwise silently disable deduplication
(lookups treat NotFound as a cache miss). An injected client is also
used to apply the initial schema mutations; when no client is
provided, table preparation keeps using its own temporary client and
behavior is unchanged.

Author: eperrine-ant

storage/gcp/antispam/gcp.go

@@ -84,6 +84,11 @@ type AntispamOpts struct {
 	// underscores, and prepending a letter if the origin doesn't start with one), to
 	// make it easy to associate tables with specific logs.
 	SpannerTablePrefix string
+
+	// SpannerClient will be used to interact with Spanner. If unset, Tessera will create one.
+	// If set, it must be connected to the database identified by the spannerDB parameter to
+	// NewAntispam, and it will never be closed by Tessera.
+	SpannerClient *spanner.Client
 }
 
 // tablePrefixRE matches valid values for a Spanner table prefix: empty, or a leading
@@ -107,11 +112,17 @@ func NewAntispam(ctx context.Context, spannerDB string, opts AntispamOpts) (*Ant
 	if !tablePrefixRE.MatchString(opts.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", opts.SpannerTablePrefix)
 	}
+	if opts.SpannerClient != nil {
+		if got := opts.SpannerClient.DatabaseName(); got != spannerDB {
+			return nil, fmt.Errorf("provided SpannerClient is connected to %q, want %q", got, spannerDB)
+		}
+	}
 	table := func(t string) string {
 		return opts.SpannerTablePrefix + t
 	}
+
 	if err := createAndPrepareTables(
-		ctx, spannerDB,
+		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")),
@@ -123,9 +134,13 @@ func NewAntispam(ctx context.Context, spannerDB string, opts AntispamOpts) (*Ant
 		return nil, fmt.Errorf("failed to create tables: %v", err)
 	}
 
-	db, err := spanner.NewClient(ctx, spannerDB)
-	if err != nil {
-		return nil, fmt.Errorf("failed to connect to Spanner: %v", err)
+	db := opts.SpannerClient
+	if db == nil {
+		var err error
+		db, err = spanner.NewClient(ctx, spannerDB)
+		if err != nil {
+			return nil, fmt.Errorf("failed to connect to Spanner: %v", err)
+		}
 	}
 
 	r := &AntispamStorage{
@@ -477,7 +492,9 @@ func (f *follower) EntriesProcessed(ctx context.Context) (uint64, error) {
 // 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, mutations [][]*spanner.Mutation) error {
+// 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 {
 		return err
@@ -499,11 +516,13 @@ func createAndPrepareTables(ctx context.Context, spannerDB string, ddl []string,
 		return err
 	}
 
-	dbPool, err := spanner.NewClient(ctx, spannerDB)
-	if err != nil {
-		return fmt.Errorf("failed to connect to Spanner: %v", 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()
 	}
-	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.

storage/gcp/antispam/gcp_test.go

@@ -24,6 +24,7 @@ import (
 
 	"log/slog"
 
+	"cloud.google.com/go/spanner"
 	"cloud.google.com/go/spanner/spannertest"
 	"github.com/transparency-dev/tessera"
 	"github.com/transparency-dev/tessera/api"
@@ -39,6 +40,7 @@ func TestAntispamStorage(t *testing.T) {
 	for _, test := range []struct {
 		name          string
 		opts          AntispamOpts
+		sharedClient  bool
 		logEntries    [][]byte
 		lookupEntries []testLookup
 	}{
@@ -62,6 +64,25 @@ func TestAntispamStorage(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:         "roundtrip with shared client",
+			opts:         AntispamOpts{SpannerTablePrefix: "Shared1_"},
+			sharedClient: true,
+			logEntries: [][]byte{
+				[]byte("one"),
+				[]byte("two"),
+			},
+			lookupEntries: []testLookup{
+				{
+					entryHash: testIDHash([]byte("one")),
+				}, {
+					entryHash: testIDHash([]byte("two")),
+				}, {
+					entryHash:    testIDHash([]byte("nowhere to be found")),
+					wantNotFound: true,
+				},
+			},
+		},
 		{
 			name: "roundtrip with table prefix",
 			opts: AntispamOpts{SpannerTablePrefix: "Tenant1_"},
@@ -84,7 +105,19 @@ func TestAntispamStorage(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)
+			const spannerDB = "projects/p/instances/i/databases/d"
+			if test.sharedClient {
+				c, err := spanner.NewClient(t.Context(), spannerDB)
+				if err != nil {
+					t.Fatalf("spanner.NewClient: %v", err)
+				}
+				// As documented on AntispamOpts.SpannerClient, the caller retains
+				// ownership of the client's lifecycle. Cleanup runs after this
+				// function's defers have shut down the follower.
+				t.Cleanup(c.Close)
+				test.opts.SpannerClient = c
+			}
+			as, err := NewAntispam(t.Context(), spannerDB, test.opts)
 			if err != nil {
 				t.Fatalf("NewAntispam: %v", err)
 			}
@@ -151,6 +184,21 @@ func TestAntispamStorage(t *testing.T) {
 	}
 }
 
+func TestAntispamSharedClientWrongDatabase(t *testing.T) {
+	closeDB := newSpannerDB(t)
+	defer closeDB()
+
+	c, err := spanner.NewClient(t.Context(), "projects/p/instances/i/databases/other")
+	if err != nil {
+		t.Fatalf("spanner.NewClient: %v", err)
+	}
+	defer c.Close()
+
+	if _, err := NewAntispam(t.Context(), "projects/p/instances/i/databases/d", AntispamOpts{SpannerClient: c}); err == nil {
+		t.Error("NewAntispam accepted a SpannerClient connected to a different database, want error")
+	}
+}
+
 func TestAntispamPushbackRecovers(t *testing.T) {
 	for _, test := range []struct {
 		name       string