perf(postgres): install notification triggers lazily on first Subscribe (#1935)

Signed-off-by: Evan <evanyan@sign.global>

Author: EvanYan1024

token/services/storage/db/sql/postgres/driver.go

@@ -106,6 +106,9 @@ func newTokenStoreProvider(dbProvider fscPostgres.DbProvider, tableNamesConfig c
 		if err != nil {
 			return nil, err
 		}
+		if o.SkipCreateTable {
+			notifier.skipSchemaManagement()
+		}
 
 		// db
 		p, err := NewTokenStoreWithNotifier(dbs, tableNames, notifier)
@@ -116,9 +119,6 @@ func newTokenStoreProvider(dbProvider fscPostgres.DbProvider, tableNamesConfig c
 			if err := p.CreateSchema(); err != nil {
 				return nil, err
 			}
-			if err := notifier.CreateSchema(); err != nil {
-				return nil, err
-			}
 		}
 
 		return p, nil
@@ -151,20 +151,14 @@ func newIdentityStoreProvider(dbProvider fscPostgres.DbProvider, tableNamesConfi
 		if err != nil {
 			return nil, err
 		}
-
-		// Get notifier for schema creation
-		notifier, err := NewIdentityNotifier(dbs, tableNames, o.DataSource)
-		if err != nil {
-			return nil, err
+		if o.SkipCreateTable {
+			p.notifier.skipSchemaManagement()
 		}
 
 		if !o.SkipCreateTable {
 			if err := p.CreateSchema(); err != nil {
 				return nil, err
 			}
-			if err := notifier.CreateSchema(); err != nil {
-				return nil, err
-			}
 		}
 
 		return p, nil
@@ -197,6 +191,9 @@ func newTransactionStoreProvider(dbProvider fscPostgres.DbProvider, tableNamesCo
 		if err != nil {
 			return nil, err
 		}
+		if o.SkipCreateTable {
+			notifier.skipSchemaManagement()
+		}
 
 		// db
 		p, err := NewTransactionStoreWithNotifier(dbs, tableNames, notifier)
@@ -207,9 +204,6 @@ func newTransactionStoreProvider(dbProvider fscPostgres.DbProvider, tableNamesCo
 			if err := p.CreateSchema(); err != nil {
 				return nil, err
 			}
-			if err := notifier.CreateSchema(); err != nil {
-				return nil, err
-			}
 		}
 
 		return p, nil

token/services/storage/db/sql/postgres/identity.go

@@ -21,8 +21,9 @@ import (
 // IdentityStore wraps common.IdentityStore to add advisory lock to schema creation
 type IdentityStore struct {
 	*sqlcommon.IdentityStore
-	writeDB *sql.DB
-	lockID  int64
+	writeDB  *sql.DB
+	lockID   int64
+	notifier *IdentityNotifier
 }
 
 // GetSchema overrides the base GetSchema to prefix with advisory lock
@@ -62,6 +63,7 @@ func NewIdentityStore(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, dataSo
 		IdentityStore: baseStore,
 		writeDB:       dbs.WriteDB,
 		lockID:        createTableLockID("identity"),
+		notifier:      notifier,
 	}, nil
 }
 

token/services/storage/db/sql/postgres/notifier.go

@@ -73,6 +73,14 @@ type Notifier struct {
 	// channelName is the name of the channel on which to receive notifications.
 	// It must be smaller than 63 characters per Postgres limit
 	channelName string
+	// ensureSchema installs the notification trigger on first subscription.
+	// Set by NewNotifier to CreateSchema; nil skips all runtime DDL and the
+	// deployment must pre-create the trigger (SkipCreateTable).
+	ensureSchema func() error
+	// startupErr records a lazy-startup failure: the trigger installation
+	// failed or the notifier was closed mid-install. Written once inside
+	// startOnce; the failure is final and every Subscribe returns it.
+	startupErr error
 }
 
 var logger = logging.MustGetLogger()
@@ -139,6 +147,7 @@ func NewNotifier(
 		closed:           false,
 		channelName:      channelName,
 	}
+	n.ensureSchema = n.CreateSchema
 
 	// attach handler that calls the subscribers
 	n.listener.Handle(channelName, &notificationHandler{
@@ -191,6 +200,23 @@ func (db *Notifier) Subscribe(callback driver.TriggerCallback) error {
 	db.startOnce.Do(func() {
 		justStarted = true
 		logger.Debugf("First subscription for notifier of [%s]. Notifier starts listening...", db.table)
+		// The notification trigger is installed on first subscription rather
+		// than at store creation: tables nobody subscribes to must not pay
+		// the per-row pg_notify cost (NOTIFY serializes transaction commits
+		// on a global queue lock).
+		if db.ensureSchema != nil {
+			if err := db.ensureSchema(); err != nil {
+				db.startupErr = errors.Wrapf(err, "failed creating notification schema for [%s]", db.table)
+
+				return
+			}
+		}
+		// the notifier may have been closed while the schema was installing
+		if err := db.ctx.Err(); err != nil {
+			db.startupErr = err
+
+			return
+		}
 		db.listenerWg.Go(func() {
 			if err := db.listener.Listen(db.ctx); err != nil {
 				// Send error to both the error channel and log it
@@ -204,6 +230,12 @@ func (db *Notifier) Subscribe(callback driver.TriggerCallback) error {
 		})
 	})
 
+	// startOnce.Do guarantees the write inside the closure is visible here.
+	// A startup failure is final: every subscription returns the same error.
+	if db.startupErr != nil {
+		return db.startupErr
+	}
+
 	if justStarted {
 		// Wait a bit to see if it fails immediately
 		timer := time.NewTimer(100 * time.Millisecond)
@@ -328,6 +360,12 @@ func (db *Notifier) GetSchema() string {
 	)
 }
 
+// skipSchemaManagement disables the lazy trigger installation; the deployment
+// is expected to pre-create the notification schema.
+func (db *Notifier) skipSchemaManagement() {
+	db.ensureSchema = nil
+}
+
 // CreateSchema creates the notification objects in the database.
 // It returns an error if the schema creation fails.
 func (db *Notifier) CreateSchema() error {