fix(recovery): address review, and fix cleanup the same way

Keystore cleanup had the same bug, so it is fixed here rather than left for a
follow-up: the id now comes from the token SKI cleanup table name and
AcquireCleanupLeadership drops its lockID param, same as recovery.

Also from review:

- deleted NewAdvisoryLockFactory and NewCleanupLeaderFactory, both had no
  production callers left and keeping them kept the caller-supplied lock id
  around, which is the footgun being removed
- the lock id tests now call the production recoveryLockID instead of
  mirroring the derivation, so they fail if it changes
- added a test that LoadConfig still succeeds with a stale advisoryLockID key
- dropped the cleanup advisoryLockID from docs/configuration.md

Signed-off-by: atharrva01 <atharvaborade568@gmail.com>

Author: atharrva01

docs/configuration.md

@@ -285,14 +285,6 @@ token:
             # Performance impact: More workers increase CPU utilization during cleanup sweeps.
             workerCount: 1
 
-            # advisoryLockID is the PostgreSQL advisory lock identifier used for cleanup leader election.
-            # This ensures only one replica performs cleanup sweeps at a time in multi-instance deployments.
-            # Default: 8389190333894887277 (hex: 0x74746b636c65616e, ASCII: "ttkclean")
-            # The default value is derived from the ASCII encoding of "ttkclean" (Token Transaction Keystore Cleanup).
-            # Only change this if you need to run multiple independent cleanup managers on the same database.
-            # Note: PostgreSQL advisory locks use 64-bit integers. This value must be unique across your application.
-            advisoryLockID: 8389190333894887277
-            
             # instanceID identifies this replica in logs and monitoring.
             # If empty, a unique identifier is generated automatically at startup.
             # Set this explicitly in containerized environments for consistent identity across restarts.
@@ -676,7 +668,6 @@ token:
             scanInterval: 1h
             batchSize: 100
             workerCount: 1
-            advisoryLockID: 8389190333894887277
             instanceID:
 ```
 
@@ -687,15 +678,14 @@ Default values:
 - scanInterval: 1h
 - batchSize: 100
 - workerCount: 1
-- advisoryLockID: 8389190333894887277 (`0x74746b636c65616e`)
 - instanceID: empty, auto-generated when the cleanup manager starts
 
 **Parameter Relationships and Tuning:**
 
 - **Cleanup is disabled by default** and must be explicitly enabled. This is a conservative default to prevent unexpected key deletion in existing deployments.
 - **Only deleted tokens older than `ttl` are considered for cleanup** to ensure tokens are truly finalized before key deletion.
 - **The manager validates** that `ttl`, `scanInterval`, `batchSize`, and `workerCount` are all greater than zero.
-- **`advisoryLockID`** is used to acquire PostgreSQL advisory-lock leadership so that only one replica performs a cleanup sweep at a time. The default value (8389190333894887277 or 0x74746b636c65616e) represents the ASCII string "ttkclean" (Token Transaction Keystore Cleanup) encoded as a 64-bit integer.
+- **Cleanup leadership** uses a PostgreSQL advisory lock so only one replica sweeps at a time. The identifier is not configurable: it is derived from the TMS's own token SKI cleanup table name, which already carries network, channel and namespace, so it is unique per TMS automatically.
 - **`instanceID`** is used to identify this replica in logs and monitoring; if omitted, the manager generates a unique identifier automatically at startup.
 
 **Tuning Recommendations:**
@@ -717,7 +707,6 @@ Default values:
 
 4. **For Multi-Instance Deployments:**
    - **PostgreSQL Required**: Multi-instance deployments require PostgreSQL for distributed coordination via advisory locks
-   - Keep default `advisoryLockID` unless running multiple independent cleanup systems
    - Consider setting explicit `instanceID` values for easier debugging and monitoring
 
 5. **For Single-Node Deployments:**
@@ -794,7 +783,6 @@ Default values:
 
 4. **For Multi-Instance Deployments:**
    - **PostgreSQL Required**: Multi-instance deployments require PostgreSQL for distributed coordination via advisory locks
-   - Keep default `advisoryLockID` unless running multiple independent recovery systems
    - Consider setting explicit `instanceID` values for easier debugging and monitoring
    - Ensure all instances share the same PostgreSQL database for proper coordination
 

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

@@ -287,7 +287,7 @@ type TokenStore interface {
 	// Returns (leadership, true, nil) if leadership was acquired.
 	// Returns (nil, false, nil) if leadership is held by another instance.
 	// Returns (nil, false, error) if an error occurred.
-	AcquireCleanupLeadership(ctx context.Context, lockID int64) (CleanupLeadership, bool, error)
+	AcquireCleanupLeadership(ctx context.Context) (CleanupLeadership, bool, error)
 }
 
 type (

token/services/storage/db/sql/common/tokens.go

@@ -47,7 +47,7 @@ type TokenStore struct {
 	table                tokenTables
 	ci                   common3.CondInterpreter
 	notifier             driver.TokenNotifier
-	cleanupLeaderFactory func(context.Context, *sql.DB, int64) (driver.CleanupLeadership, bool, error)
+	cleanupLeaderFactory func(context.Context, *sql.DB) (driver.CleanupLeadership, bool, error)
 
 	sttMutex              sync.RWMutex
 	supportedTokenFormats []token.Format
@@ -68,7 +68,7 @@ type TokenStore struct {
 	balanceStmts PreparedStmtHolder[string]
 }
 
-func newTokenStore(readDB, writeDB *sql.DB, tables tokenTables, ci common3.CondInterpreter, notifier driver.TokenNotifier, cleanupLeaderFactory func(context.Context, *sql.DB, int64) (driver.CleanupLeadership, bool, error)) *TokenStore {
+func newTokenStore(readDB, writeDB *sql.DB, tables tokenTables, ci common3.CondInterpreter, notifier driver.TokenNotifier, cleanupLeaderFactory func(context.Context, *sql.DB) (driver.CleanupLeadership, bool, error)) *TokenStore {
 	ts := &TokenStore{
 		readDB:               readDB,
 		writeDB:              writeDB,
@@ -100,7 +100,7 @@ func NewTokenStoreWithNotifierAndCleanup(
 	tables TableNames,
 	ci common3.CondInterpreter,
 	notifier driver.TokenNotifier,
-	cleanupLeaderFactory func(context.Context, *sql.DB, int64) (driver.CleanupLeadership, bool, error),
+	cleanupLeaderFactory func(context.Context, *sql.DB) (driver.CleanupLeadership, bool, error),
 ) (*TokenStore, error) {
 	return newTokenStore(readDB, writeDB, tokenTables{
 		Tokens:           tables.Tokens,
@@ -1359,12 +1359,12 @@ func (db *TokenStore) MarkTokenCleaned(ctx context.Context, txID string, index u
 // In distributed deployments (PostgreSQL), this uses advisory locks to ensure only one instance performs cleanup.
 // AcquireCleanupLeadership returns a leadership handle for cleanup sweeping.
 // When no leader factory is configured, leadership is granted locally.
-func (db *TokenStore) AcquireCleanupLeadership(ctx context.Context, lockID int64) (driver.CleanupLeadership, bool, error) {
+func (db *TokenStore) AcquireCleanupLeadership(ctx context.Context) (driver.CleanupLeadership, bool, error) {
 	if db.cleanupLeaderFactory == nil {
 		return noopCleanupLeadership{}, true, nil
 	}
 
-	return db.cleanupLeaderFactory(ctx, db.writeDB, lockID)
+	return db.cleanupLeaderFactory(ctx, db.writeDB)
 }
 
 // noopCleanupLeadership is a no-op implementation for non-distributed deployments

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

@@ -12,6 +12,7 @@ import (
 
 	"github.com/LFDT-Panurus/panurus/token/services/logging"
 	tokensdriver "github.com/LFDT-Panurus/panurus/token/services/storage/db/driver"
+	common5 "github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/common"
 	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/utils"
 )
@@ -103,16 +104,12 @@ func (l *AdvisoryLock) Close() error {
 	return nil
 }
 
-// NewAdvisoryLockFactory returns a recovery leader factory function that uses PostgreSQL advisory locks.
-func NewAdvisoryLockFactory() func(context.Context, *sql.DB, int64) (tokensdriver.RecoveryLeadership, bool, error) {
-	return func(ctx context.Context, db *sql.DB, lockID int64) (tokensdriver.RecoveryLeadership, bool, error) {
-		lock, acquired, err := NewAdvisoryLock(ctx, db, lockID)
-		if err != nil || !acquired {
-			return nil, acquired, err
-		}
-
-		return lock, true, nil
-	}
+// recoveryLockID derives the advisory lock id guarding a TMS's recovery sweep from its fully
+// qualified requests table name, which already carries the network, channel and namespace. The
+// table prefix alone is not enough: TMSes normally share one persistence configuration and are
+// distinguished only by those params, so a prefix-derived id would collide across them.
+func recoveryLockID(tables common5.TableNames) int64 {
+	return createTableLockID(tables.Requests + "_recovery")
 }
 
 // NewAdvisoryLockFactoryForID returns a recovery leader factory bound to lockID. Binding the id
@@ -129,9 +126,17 @@ func NewAdvisoryLockFactoryForID(lockID int64) func(context.Context, *sql.DB) (t
 	}
 }
 
-// NewCleanupLeaderFactory returns a cleanup leader factory function that uses PostgreSQL advisory locks.
-func NewCleanupLeaderFactory() func(context.Context, *sql.DB, int64) (tokensdriver.CleanupLeadership, bool, error) {
-	return func(ctx context.Context, db *sql.DB, lockID int64) (tokensdriver.CleanupLeadership, bool, error) {
+// keystoreCleanupLockID derives the advisory lock id guarding a TMS's keystore cleanup sweep from
+// its fully qualified token SKI cleanup table name, for the same reason as recoveryLockID: the
+// table prefix alone is shared by every TMS on one persistence configuration.
+func keystoreCleanupLockID(tables common5.TableNames) int64 {
+	return createTableLockID(tables.TokenSKICleanups + "_cleanup")
+}
+
+// NewCleanupLeaderFactoryForID returns a cleanup leader factory bound to lockID, so the id is
+// fixed when the store is built rather than supplied per call.
+func NewCleanupLeaderFactoryForID(lockID int64) func(context.Context, *sql.DB) (tokensdriver.CleanupLeadership, bool, error) {
+	return func(ctx context.Context, db *sql.DB) (tokensdriver.CleanupLeadership, bool, error) {
 		lock, acquired, err := NewAdvisoryLock(ctx, db, lockID)
 		if err != nil || !acquired {
 			return nil, acquired, err

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

@@ -164,20 +164,19 @@ func TestAdvisoryLockFactory(t *testing.T) {
 	defer utils.IgnoreErrorFunc(db.Close)
 
 	ctx := context.Background()
-	lockID := int64(77777)
 
-	// Test factory function
-	factory := NewAdvisoryLockFactory()
+	// The factory binds its lock id at construction, so callers cannot pass one in.
+	factory := NewAdvisoryLockFactoryForID(int64(77777))
 	require.NotNil(t, factory)
 
 	// Use factory to create lock
-	lock, acquired, err := factory(ctx, db, lockID)
+	lock, acquired, err := factory(ctx, db)
 	require.NoError(t, err)
 	require.True(t, acquired)
 	require.NotNil(t, lock)
 
 	// Verify it's actually locked
-	lock2, acquired, err := factory(ctx, db, lockID)
+	lock2, acquired, err := factory(ctx, db)
 	require.NoError(t, err)
 	require.False(t, acquired)
 	require.Nil(t, lock2)

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

@@ -23,11 +23,6 @@ func tmsTables(t *testing.T, prefix, network, channel, namespace string) sqlcomm
 	return tables
 }
 
-// recoveryLockID mirrors the derivation in NewTransactionStoreWithNotifier.
-func recoveryLockID(tables sqlcommon.TableNames) int64 {
-	return createTableLockID(tables.Requests + "_recovery")
-}
-
 // Two TMSes normally share one persistence configuration and differ only by the network,
 // channel and namespace params, so the recovery lock id has to be derived from something that
 // carries them. Deriving from the table prefix alone would give both the same id, and every

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

@@ -73,7 +73,7 @@ func (n *TokenNotifier) Subscribe(callback func(tokensdriver.Operation, tokensdr
 
 func NewTokenStoreWithNotifier(dbs *scommon.RWDB, tableNames sqlcommon.TableNames, notifier *TokenNotifier) (*TokenStore, error) {
 	// Create cleanup leader factory using PostgreSQL advisory locks
-	cleanupLeaderFactory := NewCleanupLeaderFactory()
+	cleanupLeaderFactory := NewCleanupLeaderFactoryForID(keystoreCleanupLockID(tableNames))
 
 	baseStore, err := sqlcommon.NewTokenStoreWithNotifierAndCleanup(
 		dbs.ReadDB,

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

@@ -74,7 +74,7 @@ func NewTransactionStoreWithNotifier(dbs *scommon.RWDB, tableNames sqlcommon.Tab
 	// the table prefix would not be enough: several TMSes normally share one persistence
 	// configuration and are distinguished only by those params, so they would all land on the same
 	// id and every TMS but one would skip its sweep. See #1798 for the same bug on the token locks.
-	recoveryLeaderFactory := NewAdvisoryLockFactoryForID(createTableLockID(tableNames.Requests + "_recovery"))
+	recoveryLeaderFactory := NewAdvisoryLockFactoryForID(recoveryLockID(tableNames))
 
 	commonStore, err := sqlcommon.NewTransactionStoreWithNotifierAndRecovery(
 		dbs.ReadDB,

token/services/storage/services/cleanup/config.go

@@ -29,25 +29,18 @@ type Config struct {
 	BatchSize int
 	// WorkerCount is the number of local workers processing tokens
 	WorkerCount int
-	// AdvisoryLockID is the PostgreSQL advisory lock ID used for leader election
-	AdvisoryLockID int64
 	// InstanceID identifies the current replica as the cleanup owner
 	InstanceID string
 }
 
-const (
-	defaultLockID int64 = 0x74746b636c65616e // "ttkclean" in hex
-)
-
 // DefaultConfig returns the default cleanup configuration
 func DefaultConfig() Config {
 	return Config{
-		Enabled:        false,          // Disabled by default - must be explicitly enabled
-		TTL:            24 * time.Hour, // Wait 24 hours before cleaning up keys
-		ScanInterval:   1 * time.Hour,  // Scan every hour
-		BatchSize:      100,
-		WorkerCount:    1,
-		AdvisoryLockID: defaultLockID,
+		Enabled:      false,          // Disabled by default - must be explicitly enabled
+		TTL:          24 * time.Hour, // Wait 24 hours before cleaning up keys
+		ScanInterval: 1 * time.Hour,  // Scan every hour
+		BatchSize:    100,
+		WorkerCount:  1,
 	}
 }
 
@@ -81,9 +74,6 @@ func LoadConfig(cfg *config.Configuration) (Config, error) {
 	if config.WorkerCount > 0 {
 		result.WorkerCount = config.WorkerCount
 	}
-	if config.AdvisoryLockID != 0 {
-		result.AdvisoryLockID = config.AdvisoryLockID
-	}
 	if config.InstanceID != "" {
 		result.InstanceID = config.InstanceID
 	}

token/services/storage/services/cleanup/manager.go

@@ -31,7 +31,7 @@ type DeletedToken struct {
 // Storage defines the interface for querying deleted tokens
 type Storage interface {
 	// AcquireCleanupLeadership acquires an advisory lock for cleanup leadership
-	AcquireCleanupLeadership(ctx context.Context, lockID int64) (Leadership, bool, error)
+	AcquireCleanupLeadership(ctx context.Context) (Leadership, bool, error)
 	// GetDeletedTokensPendingSKICleanup returns deleted tokens older than the specified duration that haven't had their SKI keys cleaned.
 	// Only tokens owned by this node are returned, since this node only holds the secret keys for tokens it owns.
 	GetDeletedTokensPendingSKICleanup(ctx context.Context, olderThan time.Duration, limit int) ([]DeletedToken, error)
@@ -139,8 +139,8 @@ func (m *Manager) Start() error {
 	m.wg.Add(1)
 	go m.cleanupLoop()
 
-	m.logger.Infof("keystore cleanup manager started (TTL: %s, Scan Interval: %s, Batch Size: %d, Workers: %d, Lock ID: %d, Instance ID: %s)",
-		m.config.TTL, m.config.ScanInterval, m.config.BatchSize, m.config.WorkerCount, m.config.AdvisoryLockID, m.config.InstanceID)
+	m.logger.Infof("keystore cleanup manager started (TTL: %s, Scan Interval: %s, Batch Size: %d, Workers: %d, Instance ID: %s)",
+		m.config.TTL, m.config.ScanInterval, m.config.BatchSize, m.config.WorkerCount, m.config.InstanceID)
 
 	return nil
 }
@@ -205,7 +205,7 @@ func (m *Manager) validateConfig() error {
 }
 
 func (m *Manager) runSweep(ctx context.Context) error {
-	leadership, acquired, err := m.storage.AcquireCleanupLeadership(ctx, m.config.AdvisoryLockID)
+	leadership, acquired, err := m.storage.AcquireCleanupLeadership(ctx)
 	if err != nil {
 		return errors.Wrapf(err, "failed to acquire cleanup leadership")
 	}