Fix: Hold Leader Election

[hayesZach]

Holds a leader election to determine if the running machine can start ingesting CSFloat reversals.
Closes CSF-1174

---

Note

Medium Risk
Introduces Postgres advisory-lock leader election to gate ingestion work, changing runtime scheduling and database interaction; misconfiguration could prevent ingestion or cause unexpected lock contention.

Overview
Adds a leader-election mechanism using Postgres advisory locks and wires it into the CSFloat ingestor so only the elected leader performs periodic sync work.

Refactors ingestors to use new domain-level Ingestor/Manager interfaces, updates the manager lifecycle (StopIngestors), and extends the repository Factory with PublicDB/PrivateDB access to support the elector; includes comprehensive tests for lock acquisition and single-leader execution.

^{Written by Cursor Bugbot for commit 94cd342. This will update automatically on new commits. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

Autofix Details

Bugbot Autofix prepared fixes for both issues found in the latest run.

• ✅ Fixed: Transaction-scoped advisory lock held open indefinitely
  • Leader election now acquires a session-scoped advisory lock on a dedicated public DB connection and no longer keeps a transaction open during the long-running leader loop.
• ✅ Fixed: Missing validation for Elector ID and Salt
  • Config loading now panics when elector mode is enabled but either Elector.ID or Elector.Salt is empty.

Or push these changes by commenting:

@cursor push 1902b6944e

Preview (1902b6944e)

diff --git a/config/config.go b/config/config.go
--- a/config/config.go
+++ b/config/config.go
@@ -120,6 +120,12 @@
 		}
 	}
 
+	if cfg.Elector.Enable {
+		if cfg.Elector.ID == "" || cfg.Elector.Salt == "" {
+			panic("elector configuration is required when enabled")
+		}
+	}
+
 	if cfg.Elector.Enable != cfg.Ingestors.CSFloat.Enable {
 		panic("both elector and ingestor require the same enable state")
 	}

diff --git a/domain/repository/factory.go b/domain/repository/factory.go
--- a/domain/repository/factory.go
+++ b/domain/repository/factory.go
@@ -1,6 +1,7 @@
 package repository
 
 import (
+	"context"
 	"io"
 
 	"gorm.io/gorm"
@@ -18,6 +19,12 @@
 	TryAdvisoryXactLock(id, salt string) (bool, error)
 }
 
+type AdvisoryLockSession interface {
+	io.Closer
+	TryAdvisoryLock(id, salt string) (bool, error)
+	AdvisoryUnlock(id, salt string) error
+}
+
 type PublicTransaction interface {
 	gorm.TxCommitter
 
@@ -38,4 +45,5 @@
 
 	NewPublicTransaction() PublicTransaction
 	RunInTransactionPublic(fn func(PublicTransaction) error) error
+	NewPublicAdvisoryLockSession(ctx context.Context) (AdvisoryLockSession, error)
 }

diff --git a/leader/leader.go b/leader/leader.go
--- a/leader/leader.go
+++ b/leader/leader.go
@@ -41,24 +41,34 @@
 		case <-time.After(time.Until(nextMinute)):
 		}
 
-		tx := e.factory.NewPublicTransaction()
-		acquired, err := tx.TryAdvisoryXactLock(e.cfg.Elector.ID, e.cfg.Elector.Salt)
+		lockSession, err := e.factory.NewPublicAdvisoryLockSession(ctx)
 		if err != nil {
+			e.log.Warnf("failed to create advisory lock session: %v", err)
+			continue
+		}
+
+		acquired, err := lockSession.TryAdvisoryLock(e.cfg.Elector.ID, e.cfg.Elector.Salt)
+		if err != nil {
 			e.log.Warnf("failed to acquire leader lock: %v", err)
-			tx.Rollback()
+			lockSession.Close()
 			continue
 		}
 
 		if !acquired {
 			e.log.Warn("failed to acquire leader lock")
-			tx.Rollback()
+			lockSession.Close()
 			continue
 		}
 
 		e.log.Info("leader lock acquired")
 		onLeader()
 
-		tx.Rollback()
+		if err := lockSession.AdvisoryUnlock(e.cfg.Elector.ID, e.cfg.Elector.Salt); err != nil {
+			e.log.Warnf("failed to release leader lock: %v", err)
+		}
+		if err := lockSession.Close(); err != nil {
+			e.log.Warnf("failed to close advisory lock session: %v", err)
+		}
 		return
 	}
 }

diff --git a/repository/factory/factory.go b/repository/factory/factory.go
--- a/repository/factory/factory.go
+++ b/repository/factory/factory.go
@@ -1,6 +1,7 @@
 package factory
 
 import (
+	"context"
 	"database/sql"
 	"errors"
 	"fmt"
@@ -210,6 +211,20 @@
 	return newPublicTransaction(f.public.Begin())
 }
 
+func (f *factory) NewPublicAdvisoryLockSession(ctx context.Context) (repository.AdvisoryLockSession, error) {
+	sqlDB, err := f.public.DB()
+	if err != nil {
+		return nil, err
+	}
+
+	conn, err := sqlDB.Conn(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return newPublicAdvisoryLockSession(conn), nil
+}
+
 func (f *factory) RunInTransactionPublic(fn func(repository.PublicTransaction) error) error {
 	return f.public.Transaction(func(gormTx *gorm.DB) error {
 		tx := newPublicTransaction(gormTx)

diff --git a/repository/factory/transaction.go b/repository/factory/transaction.go
--- a/repository/factory/transaction.go
+++ b/repository/factory/transaction.go
@@ -1,6 +1,8 @@
 package factory
 
 import (
+	"context"
+	"database/sql"
 	"hash/fnv"
 
 	"reverse-watch/domain/repository"
@@ -72,13 +74,9 @@
 }
 
 func (t *publicTransaction) TryAdvisoryXactLock(id, salt string) (bool, error) {
-	h := fnv.New32a()
-	h.Write([]byte(salt))
-	h.Write([]byte(id))
-	lockKey := h.Sum32()
-
+	lockKey := advisoryLockKey(id, salt)
 	var hasLock bool
-	if err := t.tx.Raw("SELECT pg_try_advisory_xact_lock(?)", lockKey).Scan(&hasLock).Error; err != nil {
+	if err := t.tx.Raw("SELECT pg_try_advisory_lock(?)", lockKey).Scan(&hasLock).Error; err != nil {
 		return false, err
 	}
 
@@ -87,3 +85,40 @@
 	}
 	return true, nil
 }
+
+type publicAdvisoryLockSession struct {
+	conn *sql.Conn
+}
+
+func newPublicAdvisoryLockSession(conn *sql.Conn) *publicAdvisoryLockSession {
+	return &publicAdvisoryLockSession{conn: conn}
+}
+
+func (s *publicAdvisoryLockSession) Close() error {
+	return s.conn.Close()
+}
+
+func (s *publicAdvisoryLockSession) TryAdvisoryLock(id, salt string) (bool, error) {
+	lockKey := advisoryLockKey(id, salt)
+	var hasLock bool
+	if err := s.conn.QueryRowContext(context.Background(), "SELECT pg_try_advisory_lock($1)", lockKey).Scan(&hasLock); err != nil {
+		return false, err
+	}
+	return hasLock, nil
+}
+
+func (s *publicAdvisoryLockSession) AdvisoryUnlock(id, salt string) error {
+	lockKey := advisoryLockKey(id, salt)
+	var unlocked bool
+	if err := s.conn.QueryRowContext(context.Background(), "SELECT pg_advisory_unlock($1)", lockKey).Scan(&unlocked); err != nil {
+		return err
+	}
+	return nil
+}
+
+func advisoryLockKey(id, salt string) uint32 {
+	h := fnv.New32a()
+	h.Write([]byte(salt))
+	h.Write([]byte(id))
+	return h.Sum32()
+}

_{This Bugbot Autofix run was free. To enable autofix for future PRs, go to the Cursor dashboard.}

[linear]

CSF-1174 Hold Leader Election

Hold a leader election to determine which machine will ingest CSFloat reversals. Each machine will try to acquire a leader lock every minute, indefinitely.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}