From 6cfa276cdc9b0d916ac31a1ee6f1cebd39af10d1 Mon Sep 17 00:00:00 2001 From: Glenn Rempe Date: Thu, 22 Oct 2026 18:06:15 -0400 Subject: [PATCH] storage/aws: accept a caller-owned MySQL DB The AWS driver opens its own MySQL pools, for the sequencer and for the persistent antispam, and never closes them, so every in-process reopen of a log leaks the pools' goroutines and idle connections. Add an optional DB to Config and to AntispamOpts, mirroring the GCP driver's SpannerClient options. If it is set, Tessera uses it and never closes it, and ignores the pool settings, and Config also ignores its DSN. The antispam checks that the DB is connected to the database its dsn names, since a mismatch would silently break duplicate detection. --- storage/aws/antispam/aws.go | 37 +++++++++++++++++- storage/aws/antispam/aws_test.go | 64 +++++++++++++++++++++++++++++++- storage/aws/aws.go | 22 ++++++++++- storage/aws/aws_test.go | 29 +++++++++++++++ 4 files changed, 148 insertions(+), 4 deletions(-) diff --git a/storage/aws/antispam/aws.go b/storage/aws/antispam/aws.go index 2c6332fde..6c06a33de 100644 --- a/storage/aws/antispam/aws.go +++ b/storage/aws/antispam/aws.go @@ -35,7 +35,7 @@ import ( "github.com/transparency-dev/tessera/internal/otel" "go.opentelemetry.io/otel/trace" - _ "github.com/go-sql-driver/mysql" + "github.com/go-sql-driver/mysql" ) const ( @@ -71,6 +71,11 @@ type AntispamOpts struct { // the antispam decorator will return a wrapped tessera.ErrPushback for every Add request. PushbackThreshold uint + // DB will be used to interact with MySQL. If unset, Tessera will open one using the + // dsn parameter to NewAntispam, MaxOpenConns and MaxIdleConns. If set, it must be + // connected to the database named by dsn, and it will never be closed by Tessera. + DB *sql.DB + PushbackMaxOutstanding uint64 MaxOpenConns int MaxIdleConns int @@ -107,6 +112,13 @@ func NewAntispam(ctx context.Context, dsn string, opts AntispamOpts) (*AntispamS opts.PushbackThreshold = DefaultPushbackThreshold } + if opts.DB != nil { + if err := checkDatabase(ctx, opts.DB, dsn); err != nil { + return nil, err + } + return newAntispam(ctx, opts.DB, opts) + } + dbPool, err := sql.Open("mysql", dsn) if err != nil { return nil, fmt.Errorf("failed to connect to MySQL db: %v", err) @@ -118,7 +130,12 @@ func NewAntispam(ctx context.Context, dsn string, opts AntispamOpts) (*AntispamS if opts.MaxIdleConns >= 0 { dbPool.SetMaxIdleConns(opts.MaxIdleConns) } + return newAntispam(ctx, dbPool, opts) +} +// newAntispam returns an antispam driver which uses the provided database handle +// for its MySQL connections. +func newAntispam(ctx context.Context, dbPool *sql.DB, opts AntispamOpts) (*AntispamStorage, error) { if err := dbPool.Ping(); err != nil { return nil, fmt.Errorf("failed to ping MySQL db: %v", err) } @@ -137,6 +154,24 @@ func NewAntispam(ctx context.Context, dsn string, opts AntispamOpts) (*AntispamS return r, nil } +// checkDatabase returns an error unless db is connected to the database named by dsn. +// A pool connected to another database would otherwise keep a separate, and so +// incomplete, record of the log's entries, and fail to detect duplicates. +func checkDatabase(ctx context.Context, db *sql.DB, dsn string) error { + cfg, err := mysql.ParseDSN(dsn) + if err != nil { + return fmt.Errorf("failed to parse DSN: %v", err) + } + var got sql.NullString + if err := db.QueryRowContext(ctx, "SELECT DATABASE()").Scan(&got); err != nil { + return fmt.Errorf("failed to read the database of the provided DB: %v", err) + } + if got.String != cfg.DBName { + return fmt.Errorf("provided DB is connected to database %q, want %q", got.String, cfg.DBName) + } + return nil +} + func (s *AntispamStorage) initDB(ctx context.Context) error { if _, err := s.dbPool.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS AntispamMeta ( diff --git a/storage/aws/antispam/aws_test.go b/storage/aws/antispam/aws_test.go index 3c88d0088..89ad036c2 100644 --- a/storage/aws/antispam/aws_test.go +++ b/storage/aws/antispam/aws_test.go @@ -25,6 +25,7 @@ import ( "log/slog" + "github.com/go-sql-driver/mysql" "github.com/transparency-dev/tessera" "github.com/transparency-dev/tessera/api" "github.com/transparency-dev/tessera/testonly" @@ -42,16 +43,44 @@ func TestMain(m *testing.M) { } func TestAntispam(t *testing.T) { + for _, test := range []struct { + name string + sharedDB bool + }{ + {name: "own DB"}, + {name: "shared DB", sharedDB: true}, + } { + t.Run(test.name, func(t *testing.T) { testAntispam(t, test.sharedDB) }) + } +} + +// testAntispam drives the full append, follow and duplicate lookup cycle, through a DB +// provided in AntispamOpts if sharedDB is set. +func testAntispam(t *testing.T, sharedDB bool) { ctx := t.Context() if canSkipMySQLTest(t, ctx) { slog.WarnContext(ctx, "MySQL not available, skipping", slog.String("name", t.Name())) t.Skip("MySQL not available, skipping test") } mustDropTables(t, ctx) - as, err := NewAntispam(ctx, *mySQLURI, AntispamOpts{}) + opts := AntispamOpts{} + if sharedDB { + db, err := sql.Open("mysql", *mySQLURI) + if err != nil { + t.Fatal(err) + } + // As documented on AntispamOpts.DB, the caller retains ownership of the DB's + // lifecycle. Cleanup runs after this function's defers have shut down the log. + t.Cleanup(func() { _ = db.Close() }) + opts.DB = db + } + as, err := NewAntispam(ctx, *mySQLURI, opts) if err != nil { t.Fatal(err) } + if sharedDB && as.dbPool != opts.DB { + t.Fatal("NewAntispam did not use the provided DB") + } fl, shutdown := testonly.NewTestLog(t, tessera.NewAppendOptions().WithCheckpointInterval(time.Second)) defer func() { if err := shutdown(t.Context()); err != nil { @@ -107,6 +136,39 @@ func TestAntispam(t *testing.T) { } } +func TestAntispamSharedDBWrongDatabase(t *testing.T) { + ctx := t.Context() + if canSkipMySQLTest(t, ctx) { + slog.WarnContext(ctx, "MySQL not available, skipping", slog.String("name", t.Name())) + t.Skip("MySQL not available, skipping test") + } + // A second, empty database that the provided DB is connected to instead. + admin, err := sql.Open("mysql", *mySQLURI) + if err != nil { + t.Fatal(err) + } + defer func() { _ = admin.Close() }() + const other = "test_tessera_other" + if _, err := admin.ExecContext(ctx, "CREATE DATABASE IF NOT EXISTS "+other); err != nil { + t.Fatal(err) + } + defer func() { _, _ = admin.ExecContext(context.Background(), "DROP DATABASE "+other) }() + cfg, err := mysql.ParseDSN(*mySQLURI) + if err != nil { + t.Fatal(err) + } + cfg.DBName = other + db, err := sql.Open("mysql", cfg.FormatDSN()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = db.Close() }() + + if _, err := NewAntispam(ctx, *mySQLURI, AntispamOpts{DB: db}); err == nil { + t.Error("NewAntispam accepted a DB connected to a different database, want error") + } +} + func TestAntispamPushbackRecovers(t *testing.T) { ctx := t.Context() if canSkipMySQLTest(t, ctx) { diff --git a/storage/aws/aws.go b/storage/aws/aws.go index f4c0aa26c..5beaca2af 100644 --- a/storage/aws/aws.go +++ b/storage/aws/aws.go @@ -162,6 +162,10 @@ type Config struct { MaxOpenConns int // Maximum idle database connections in the connection pool. MaxIdleConns int + // DB will be used to interact with MySQL. If unset, Tessera will open one using DSN, + // MaxOpenConns and MaxIdleConns. If set, those are ignored, and it will never be + // closed by Tessera. + DB *sql.DB // HTTPClient will be used for other HTTP requests. If unset, Tessera will use the net/http DefaultClient. HTTPClient *http.Client @@ -196,7 +200,7 @@ func New(ctx context.Context, cfg Config) (tessera.Driver, error) { // Appender creates a new tessera.Appender lifecycle object. func (s *Storage) Appender(ctx context.Context, opts *tessera.AppendOptions) (*tessera.Appender, tessera.LogReader, error) { - seq, err := newMySQLSequencer(ctx, s.cfg.DSN, uint64(opts.PushbackMaxOutstanding()), s.cfg.MaxOpenConns, s.cfg.MaxIdleConns) + seq, err := s.sequencer(ctx, uint64(opts.PushbackMaxOutstanding())) if err != nil { return nil, nil, fmt.Errorf("failed to create MySQL sequencer: %v", err) } @@ -553,7 +557,7 @@ func (s *Storage) MigrationWriter(ctx context.Context, opts *tessera.MigrationOp }, entriesPath: opts.EntriesPath(), } - seq, err := newMySQLSequencer(ctx, s.cfg.DSN, DefaultPushbackMaxOutstanding, s.cfg.MaxOpenConns, s.cfg.MaxIdleConns) + seq, err := s.sequencer(ctx, DefaultPushbackMaxOutstanding) if err != nil { return nil, nil, fmt.Errorf("failed to create MySQL sequencer: %v", err) } @@ -896,6 +900,15 @@ type mySQLSequencer struct { maxOutstanding uint64 } +// sequencer returns a mySQLSequencer which uses the configured DB, or opens one +// for the configured DSN. +func (s *Storage) sequencer(ctx context.Context, maxOutstanding uint64) (*mySQLSequencer, error) { + if s.cfg.DB != nil { + return newMySQLSequencerWithDB(ctx, s.cfg.DB, maxOutstanding) + } + return newMySQLSequencer(ctx, s.cfg.DSN, maxOutstanding, s.cfg.MaxOpenConns, s.cfg.MaxIdleConns) +} + // newMySQLSequencer returns a new mysqlSequencer struct which uses the provided // DSN for its MySQL connection. func newMySQLSequencer(ctx context.Context, dsn string, maxOutstanding uint64, maxOpenConns, maxIdleConns int) (*mySQLSequencer, error) { @@ -910,7 +923,12 @@ func newMySQLSequencer(ctx context.Context, dsn string, maxOutstanding uint64, m if maxIdleConns >= 0 { dbPool.SetMaxIdleConns(maxIdleConns) } + return newMySQLSequencerWithDB(ctx, dbPool, maxOutstanding) +} +// newMySQLSequencerWithDB returns a new mysqlSequencer struct which uses the provided +// database handle for its MySQL connections. +func newMySQLSequencerWithDB(ctx context.Context, dbPool *sql.DB, maxOutstanding uint64) (*mySQLSequencer, error) { if err := dbPool.Ping(); err != nil { return nil, fmt.Errorf("failed to ping MySQL db: %v", err) } diff --git a/storage/aws/aws_test.go b/storage/aws/aws_test.go index 1eee0ddc1..02b2a6416 100644 --- a/storage/aws/aws_test.go +++ b/storage/aws/aws_test.go @@ -210,6 +210,35 @@ func TestMySQLSequencerPushback(t *testing.T) { } } +// TestStorageUsesProvidedDB checks that Storage sequences with the DB set in its +// Config, rather than opening one for its DSN. +func TestStorageUsesProvidedDB(t *testing.T) { + ctx := context.Background() + if canSkipMySQLTest(t, ctx) { + slog.WarnContext(ctx, "MySQL not available, skipping", slog.String("name", t.Name())) + t.Skip("MySQL not available, skipping test") + } + mustDropTables(t, ctx) + db, err := sql.Open("mysql", *mySQLURI) + if err != nil { + t.Fatal(err) + } + // As documented on Config.DB, the caller retains ownership of the DB's lifecycle. + defer func() { _ = db.Close() }() + + s := &Storage{cfg: Config{DB: db, DSN: "nobody:nothing@tcp(127.0.0.1:1)/nowhere"}} + seq, err := s.sequencer(ctx, 1000) + if err != nil { + t.Fatalf("sequencer: %v", err) + } + if seq.dbPool != db { + t.Fatal("the sequencer does not use the provided DB") + } + if err := seq.assignEntries(ctx, []*tessera.Entry{tessera.NewEntry([]byte("entry"))}); err != nil { + t.Fatalf("assignEntries: %v", err) + } +} + func TestMySQLSequencerRoundTrip(t *testing.T) { ctx := context.Background() if canSkipMySQLTest(t, ctx) {