Merge pull request #60 from csfloat/fix/leader-election

Fix: Hold Leader Election

Author: zedimytch

config/config.go

@@ -110,7 +110,6 @@ func load() Config {
 			panic("csfloat ingestor configuration is required when enabled")
 		}
 	}
-
 	return cfg
 }
 

domain/ingestors/ingestors.go

@@ -0,0 +1,17 @@
+package ingestors
+
+type Ingestor interface {
+	Start()
+	Stop()
+}
+
+type Manager interface {
+	StartIngestors()
+	StopIngestors()
+}
+
+type IngestorType string
+
+const (
+	IngestorTypeCSFloat IngestorType = "csfloat"
+)

domain/leader/leader.go

@@ -0,0 +1,10 @@
+package leader
+
+import (
+	"context"
+	"time"
+)
+
+type Elector interface {
+	Run(ctx context.Context, lockKey uint32, period time.Duration, onWork func(ctx context.Context))
+}

domain/repository/factory.go

@@ -30,7 +30,9 @@ type Factory interface {
 
 	NewPrivateTransaction() PrivateTransaction
 	RunInTransactionPrivate(fn func(PrivateTransaction) error) error
+	PrivateDB() *gorm.DB
 
 	NewPublicTransaction() PublicTransaction
 	RunInTransactionPublic(fn func(PublicTransaction) error) error
+	PublicDB() *gorm.DB
 }

ingestors/csfloat/csfloat.go

@@ -4,11 +4,14 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"hash/fnv"
 	"net/http"
 	"time"
 
 	"reverse-watch/config"
 	"reverse-watch/domain/dto"
+	"reverse-watch/domain/ingestors"
+	"reverse-watch/domain/leader"
 	"reverse-watch/domain/models"
 	"reverse-watch/domain/repository"
 	"reverse-watch/errors"
@@ -21,18 +24,22 @@ type csfloatIngestor struct {
 	log     *zap.SugaredLogger
 	cfg     *config.Config
 	factory repository.Factory
+	elector leader.Elector
 
 	ctx     context.Context
 	cancel  context.CancelFunc
 	stopped chan struct{}
 }
 
-func NewCSFloatIngestor(ctx context.Context, factory repository.Factory, cfg *config.Config, logger *zap.SugaredLogger) *csfloatIngestor {
+var _ ingestors.Ingestor = (*csfloatIngestor)(nil)
+
+func NewCSFloatIngestor(ctx context.Context, factory repository.Factory, elector leader.Elector, cfg *config.Config, log *zap.SugaredLogger) ingestors.Ingestor {
 	ingestorCtx, cancel := context.WithCancel(ctx)
 	return &csfloatIngestor{
-		log:     logger,
+		log:     log,
 		cfg:     cfg,
 		factory: factory,
+		elector: elector,
 		ctx:     ingestorCtx,
 		cancel:  cancel,
 		stopped: make(chan struct{}),
@@ -56,7 +63,7 @@ type errorResponse struct {
 }
 
 // fetch reversal warnings from CSFloat
-func (i *csfloatIngestor) fetch(cursor *uint) ([]*slimWarning, *uint, error) {
+func (i *csfloatIngestor) fetch(ctx context.Context, cursor *uint) ([]*slimWarning, *uint, error) {
 	// The day before Valve introduced trade reversals
 	startTime := time.Date(2025, 7, 14, 0, 0, 0, 0, time.UTC)
 	endTime := time.Now().Add(-5 * time.Minute)
@@ -71,7 +78,7 @@ func (i *csfloatIngestor) fetch(cursor *uint) ([]*slimWarning, *uint, error) {
 		return nil, nil, err
 	}
 	r.Header.Set("X-Secret-Key", i.cfg.Ingestors.CSFloat.SecretKey)
-	r = r.WithContext(i.ctx)
+	r = r.WithContext(ctx)
 
 	client := &http.Client{}
 	resp, err := client.Do(r)
@@ -120,7 +127,7 @@ func (i *csfloatIngestor) process(warnings []*slimWarning) error {
 	return nil
 }
 
-func (i *csfloatIngestor) sync() error {
+func (i *csfloatIngestor) sync(ctx context.Context) error {
 	// Fetch the most recent reversals created by CSFloat
 	reversals, err := i.factory.Reversal().List(&dto.ReversalListOptions{
 		MarketplaceSlug: util.Ptr("csfloat"),
@@ -140,7 +147,13 @@ func (i *csfloatIngestor) sync() error {
 	}
 
 	for {
-		warnings, nextCursor, err := i.fetch(cursor)
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
+		warnings, nextCursor, err := i.fetch(ctx, cursor)
 		if err != nil {
 			return fmt.Errorf("failed to fetch warnings with cursor %v: %v", cursor, err)
 		}
@@ -161,26 +174,24 @@ func (i *csfloatIngestor) sync() error {
 }
 
 func (i *csfloatIngestor) Start() {
-	i.log.Info("Starting CSFloat ingestor")
+	i.log.Info("starting csfloat ingestor")
 	go func() {
 		defer close(i.stopped)
 
-		for {
-			if err := i.sync(); err != nil {
-				i.log.Errorf("failed to sync reversals: %v", err)
-			}
+		h := fnv.New32a()
+		h.Write([]byte("csfloat_ingestor_leader"))
+		lockKey := h.Sum32()
 
-			select {
-			case <-time.After(30 * time.Minute):
-			case <-i.ctx.Done():
-				return
+		i.elector.Run(i.ctx, lockKey, 30*time.Minute, func(ctx context.Context) {
+			if err := i.sync(ctx); err != nil {
+				i.log.Errorf("failed to sync reversals: %v", err)
 			}
-		}
+		})
 	}()
 }
 
 func (i *csfloatIngestor) Stop() {
-	i.log.Infof("Stopping CSFloat ingestor")
+	i.log.Infof("stopping csfloat ingestor")
 	i.cancel()
 	<-i.stopped
 }

ingestors/manager.go

@@ -4,36 +4,36 @@ import (
 	"context"
 
 	"reverse-watch/config"
+	"reverse-watch/domain/ingestors"
 	"reverse-watch/domain/repository"
 	"reverse-watch/ingestors/csfloat"
+	"reverse-watch/leader"
 
 	"go.uber.org/zap"
 )
 
-type ingestor interface {
-	Start()
-	Stop()
-}
-
 type manager struct {
 	log       *zap.SugaredLogger
-	ingestors []ingestor
+	ingestors map[ingestors.IngestorType]ingestors.Ingestor
 
 	ctx    context.Context
 	cancel context.CancelFunc
 }
 
-func New(factory repository.Factory, cfg *config.Config, logger *zap.SugaredLogger) *manager {
+var _ ingestors.Manager = (*manager)(nil)
+
+func New(factory repository.Factory, cfg *config.Config, log *zap.SugaredLogger) ingestors.Manager {
 	ctx, cancel := context.WithCancel(context.Background())
 	m := &manager{
-		log:       logger,
-		ingestors: make([]ingestor, 0),
+		log:       log,
+		ingestors: make(map[ingestors.IngestorType]ingestors.Ingestor),
 		ctx:       ctx,
 		cancel:    cancel,
 	}
 
+	elector := leader.New(factory, log)
 	if cfg.Ingestors.CSFloat.Enable {
-		m.ingestors = append(m.ingestors, csfloat.NewCSFloatIngestor(ctx, factory, cfg, logger))
+		m.ingestors[ingestors.IngestorTypeCSFloat] = csfloat.NewCSFloatIngestor(ctx, factory, elector, cfg, log)
 	}
 	return m
 }
@@ -44,7 +44,7 @@ func (m *manager) StartIngestors() {
 	}
 }
 
-func (m *manager) Stop() {
+func (m *manager) StopIngestors() {
 	m.cancel()
 	for _, ingestor := range m.ingestors {
 		ingestor.Stop()

leader/leader.go

@@ -0,0 +1,93 @@
+package leader
+
+import (
+	"context"
+	"database/sql"
+	"time"
+
+	"reverse-watch/domain/leader"
+	"reverse-watch/domain/repository"
+
+	"go.uber.org/zap"
+)
+
+type elector struct {
+	log     *zap.SugaredLogger
+	factory repository.Factory
+}
+
+var _ leader.Elector = (*elector)(nil)
+
+func New(factory repository.Factory, log *zap.SugaredLogger) leader.Elector {
+	return &elector{
+		log:     log,
+		factory: factory,
+	}
+}
+
+func (e *elector) tryAdvisoryLock(ctx context.Context, lockKey uint32) (*sql.Tx, bool) {
+	db, err := e.factory.PublicDB().DB()
+	if err != nil {
+		e.log.Errorf("failed to get database connection: %v", err)
+		return nil, false
+	}
+
+	// Use background context so transaction isn't auto-cancelled.
+	// Very unlikely, but we don't want another instance to acquire the lock and begin work before we have exited.
+	tx, err := db.BeginTx(context.Background(), nil)
+	if err != nil {
+		e.log.Errorf("failed to begin transaction: %v", err)
+		return nil, false
+	}
+
+	var hasLock bool
+	err = tx.QueryRowContext(ctx, "SELECT pg_try_advisory_xact_lock($1)", lockKey).Scan(&hasLock)
+	if err != nil {
+		tx.Rollback()
+		e.log.Errorf("failed to acquire advisory lock: %v", err)
+		return nil, false
+	}
+
+	if !hasLock {
+		tx.Rollback()
+		e.log.Errorf("failed to acquire advisory lock")
+		return nil, false
+	}
+	return tx, true
+}
+
+// Run will run the elector in a loop, acquiring the leader lock and calling the onWork function when the leader is acquired.
+// If the leader lock is not acquired, it will wait for the next period and try again.
+// If the context is done, it will return.
+func (e *elector) Run(ctx context.Context, lockKey uint32, period time.Duration, onWork func(ctx context.Context)) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		func() {
+			workCtx, workCancel := context.WithCancel(ctx)
+			defer workCancel()
+			defer waitUntilNextBoundary(ctx, period)
+			tx, acquired := e.tryAdvisoryLock(workCtx, lockKey)
+			if !acquired {
+				return
+			}
+			defer tx.Rollback()
+
+			onWork(workCtx)
+		}()
+
+	}
+}
+
+func waitUntilNextBoundary(ctx context.Context, period time.Duration) {
+	nextBoundary := time.Now().Truncate(period).Add(period)
+	select {
+	case <-ctx.Done():
+		return
+	case <-time.After(time.Until(nextBoundary)):
+	}
+}