fix: data race

Author: gfyrag

deployments/pulumi/main.go

@@ -3,9 +3,11 @@ package main
 import (
 	"errors"
 	"fmt"
+	"github.com/formancehq/go-libs/v2/pointer"
 	"github.com/formancehq/ledger/deployments/pulumi/pkg"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumi/config"
+	"github.com/pulumi/pulumi/sdk/v3/go/pulumix"
 )
 
 func main() {
@@ -43,9 +45,9 @@ func deploy(ctx *pulumi.Context) error {
 		Postgres: pulumi_ledger.PostgresArgs{
 			URI: pulumi.String(postgresURI),
 		},
-		Debug:                pulumi.Bool(conf.GetBool("debug")),
+		Debug: pulumi.Bool(conf.GetBool("debug")),
 		API: pulumi_ledger.APIArgs{
-			ReplicaCount:         pulumi.Int(conf.GetInt("replicaCount")),
+			ReplicaCount:         pulumix.Val(pointer.For(conf.GetInt("replicaCount"))),
 			ExperimentalFeatures: pulumi.Bool(conf.GetBool("experimentalFeatures")),
 		},
 	})

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/antlr/antlr4/runtime/Go/antlr v1.4.10
 	github.com/bluele/gcache v0.0.2
 	github.com/dop251/goja v0.0.0-20241009100908-5f46f2705ca3
-	github.com/formancehq/go-libs/v2 v2.0.1-0.20250212194833-6c71efac350c
+	github.com/formancehq/go-libs/v2 v2.0.1-0.20250220173034-9ab923e71c39
 	github.com/formancehq/ledger/pkg/client v0.0.0-00010101000000-000000000000
 	github.com/go-chi/chi/v5 v5.2.1
 	github.com/go-chi/cors v1.2.1

go.sum

@@ -112,6 +112,8 @@ github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/formancehq/go-libs/v2 v2.0.1-0.20250212194833-6c71efac350c h1:wvCzbTbuWmhumY0h31SvExQLUzKlx3WQ7CK/Xwk8Ib0=
 github.com/formancehq/go-libs/v2 v2.0.1-0.20250212194833-6c71efac350c/go.mod h1:U9+fV6R34gG24TfNuOCUMQmfdP3g54nsFlLNlManZ5s=
+github.com/formancehq/go-libs/v2 v2.0.1-0.20250220173034-9ab923e71c39 h1:e2gh25n8pRPMYBFGuIhMR2Aa7DBwahFhVXsIFw5Mbok=
+github.com/formancehq/go-libs/v2 v2.0.1-0.20250220173034-9ab923e71c39/go.mod h1:U9+fV6R34gG24TfNuOCUMQmfdP3g54nsFlLNlManZ5s=
 github.com/formancehq/numscript v0.0.10 h1:ElvYpoayUX5tHtCCR18ihJTjNlHzdkE4M0IqSm9aufg=
 github.com/formancehq/numscript v0.0.10/go.mod h1:btuSv05cYwi9BvLRxVs5zrunU+O1vTgigG1T6UsawcY=
 github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=

internal/README.md

@@ -903,7 +903,7 @@ func (m Moves) ComputePostCommitEffectiveVolumes() PostCommitVolumes
 
 
 <a name="Pipeline"></a>
-## type [Pipeline](<https://github.com/formancehq/ledger/blob/main/internal/pipeline.go#L27-L36>)
+## type [Pipeline](<https://github.com/formancehq/ledger/blob/main/internal/pipeline.go#L27-L37>)
 
 
 
@@ -914,14 +914,15 @@ type Pipeline struct {
     PipelineConfiguration
     CreatedAt time.Time `json:"createdAt" bun:"created_at"`
     ID        string    `json:"id" bun:"id,pk"`
+    Version   int       `json:"-" bun:"version"`
     Enabled   bool      `json:"enabled" bun:"enabled"`
-    LastLogID *int      `json:"lastLogID,omitempty" bun:"last_log_id"`
+    LastLogID int       `json:"lastLogID,omitempty" bun:"last_log_id"`
     Error     string    `json:"error,omitempty" bun:"error"`
 }
 ```
 
 <a name="NewPipeline"></a>
-### func [NewPipeline](<https://github.com/formancehq/ledger/blob/main/internal/pipeline.go#L38>)
+### func [NewPipeline](<https://github.com/formancehq/ledger/blob/main/internal/pipeline.go#L39>)
 
 ```go
 func NewPipeline(pipelineConfiguration PipelineConfiguration) Pipeline

internal/pipeline.go

@@ -30,8 +30,9 @@ type Pipeline struct {
 	PipelineConfiguration
 	CreatedAt time.Time     `json:"createdAt" bun:"created_at"`
 	ID        string        `json:"id" bun:"id,pk"`
+	Version int `json:"-" bun:"version"`
 	Enabled   bool   `json:"enabled" bun:"enabled"`
-	LastLogID *int    `json:"lastLogID,omitempty" bun:"last_log_id"`
+	LastLogID int    `json:"lastLogID,omitempty" bun:"last_log_id"`
 	Error     string `json:"error,omitempty" bun:"error"`
 }
 
@@ -41,5 +42,6 @@ func NewPipeline(pipelineConfiguration PipelineConfiguration) Pipeline {
 		PipelineConfiguration: pipelineConfiguration,
 		Enabled:               true,
 		CreatedAt:             time.Now(),
+		LastLogID: -1,
 	}
 }

internal/replication/drivers/clickhouse/driver_test.go

@@ -14,7 +14,7 @@ import (
 	"testing"
 )
 
-func TestClickhouseConnector(t *testing.T) {
+func TestClickhouseDriver(t *testing.T) {
 	t.Parallel()
 
 	ctx := context.TODO()

internal/replication/runner/pipeline.go

@@ -62,16 +62,11 @@ func (p *PipelineHandler) Run(ctx context.Context, ingestedLogs chan int) {
 			close(ch)
 			return
 		case <-time.After(nextInterval):
-			id := -1
-			if p.pipeline.LastLogID != nil {
-				id = *p.pipeline.LastLogID
-			}
-
 			logs, err := p.store.ListLogs(ctx, pagination.ColumnPaginatedQuery[any]{
 				PageSize: 100,
 				Column:   "id",
 				Options: pagination.ResourceQuery[any]{
-					Builder: query.Gt("id", id),
+					Builder: query.Gt("id", p.pipeline.LastLogID),
 				},
 				Order: pointer.For(bunpaginate.Order(bunpaginate.OrderAsc)),
 			})
@@ -119,12 +114,12 @@ func (p *PipelineHandler) Run(ctx context.Context, ingestedLogs chan int) {
 
 			wg.Wait()
 
-			p.pipeline.LastLogID = logs.Data[len(logs.Data)-1].ID
+			p.pipeline.LastLogID = *logs.Data[len(logs.Data)-1].ID
 
 			select {
 			case <-ctx.Done():
 				return
-			case ingestedLogs <- *p.pipeline.LastLogID:
+			case ingestedLogs <- p.pipeline.LastLogID:
 			}
 
 			if !logs.HasMore {
@@ -176,4 +171,4 @@ func NewPipelineHandler(
 			WithField("module", pipeline.Ledger).
 			WithField("connector", pipeline.ConnectorID),
 	}
-}
+}

internal/replication/runner/runner.go

@@ -60,10 +60,20 @@ func (runner *Runner) stopConnector(ctx context.Context, connector drivers.Drive
 	}
 }
 
-func (runner *Runner) StartPipeline(ctx context.Context, pipeline ledger.Pipeline) (*PipelineHandler, error) {
+func (runner *Runner) StartPipeline(ctx context.Context, pipelineID string) (*PipelineHandler, error) {
 	runner.mu.Lock()
 	defer runner.mu.Unlock()
 
+	pipeline, err := runner.storage.GetPipeline(ctx, pipelineID)
+	if err != nil {
+		return nil, err
+	}
+
+	return runner.startPipeline(ctx, *pipeline)
+}
+
+func (runner *Runner) startPipeline(ctx context.Context, pipeline ledger.Pipeline) (*PipelineHandler, error) {
+	runner.logger.Infof("initializing pipeline")
 	_, ok := runner.pipelines[pipeline.ID]
 	if ok {
 		return nil, ledger.NewErrAlreadyStarted(pipeline.ID)
@@ -72,15 +82,15 @@ func (runner *Runner) StartPipeline(ctx context.Context, pipeline ledger.Pipelin
 	ctx = logging.ContextWithLogger(
 		ctx,
 		runner.logger.WithFields(map[string]any{
-			"module":    pipeline.Ledger,
+			"ledger":    pipeline.Ledger,
 			"connector": pipeline.ConnectorID,
 		}),
 	)
 
 	// Detach the context as once the process of pipeline initialisation is started, we must not stop it
 	ctx = context.WithoutCancel(ctx)
 
-	runner.logger.Infof("initializing pipeline")
+	runner.logger.Infof("initializing connector")
 	if err := runner.initConnector(pipeline.ConnectorID); err != nil {
 		return nil, err
 	}
@@ -103,6 +113,7 @@ func (runner *Runner) StartPipeline(ctx context.Context, pipeline ledger.Pipelin
 	// ignore the cancel function, as it will be called by the pipeline at its end
 	subscription := make(chan int)
 
+	runner.logger.Infof("starting handler")
 	go func() {
 		for lastLogID := range subscription {
 			if err := runner.storage.StorePipelineState(ctx, pipeline.ID, lastLogID); err != nil {
@@ -162,26 +173,34 @@ func (runner *Runner) stopConnectorIfNeeded(ctx context.Context, handler *Pipeli
 }
 
 func (runner *Runner) synchronizePipelines(ctx context.Context) error {
+	runner.mu.Lock()
+	defer runner.mu.Unlock()
+
 	runner.logger.Debug("restore pipelines from store")
+	defer func() {
+		runner.logger.Debug("restoration terminated")
+	}()
 	pipelines, err := runner.storage.ListEnabledPipelines(ctx)
 	if err != nil {
 		return fmt.Errorf("reading pipelines from store: %w", err)
 	}
 
 	for _, pipeline := range pipelines {
-		if handler := runner.GetPipeline(pipeline.ID); handler != nil {
-			if pipeline.LastLogID != nil ||
-				handler.pipeline.LastLogID == nil ||
-				*handler.pipeline.LastLogID == 0 {
+		runner.logger.Debugf("restoring pipeline %s", pipeline.ID)
+		if handler := runner.pipelines[pipeline.ID]; handler != nil {
+			if pipeline.Version == handler.pipeline.Version {
+				runner.logger.Debugf("pipeline %s up to date, skipping", pipeline.ID)
 				continue
 			}
 
+			runner.logger.Debugf("pipeline %s outdated, stopping it", pipeline.ID)
 			if err := runner.stopPipeline(ctx, pipeline.ID); err != nil {
 				runner.logger.Errorf("error stopping pipeline: %s", err)
 				continue
 			}
 		}
-		if _, err := runner.StartPipeline(ctx, pipeline); err != nil {
+		runner.logger.Debugf("starting pipeline %s", pipeline.ID)
+		if _, err := runner.startPipeline(ctx, pipeline); err != nil {
 			return err
 		}
 	}
@@ -203,6 +222,41 @@ l:
 	return nil
 }
 
+func (runner *Runner) Refresh(ctx context.Context, pipelineID string) error {
+	runner.mu.Lock()
+	defer runner.mu.Unlock()
+
+	pipeline, err := runner.storage.GetPipeline(ctx, pipelineID)
+	if err != nil {
+		return err
+	}
+
+	if pipeline.Enabled {
+		handler, ok := runner.pipelines[pipelineID]
+		if !ok {
+			if _, err := runner.startPipeline(ctx, *pipeline); err != nil {
+				return err
+			}
+			return nil
+		} else if pipeline.Version != handler.pipeline.Version {
+			if err := runner.stopPipeline(ctx, pipelineID); err != nil {
+				return err
+			}
+			if _, err := runner.startPipeline(ctx, *pipeline); err != nil {
+				return err
+			}
+		}
+
+	} else {
+		_, ok := runner.pipelines[pipelineID]
+		if ok {
+			return runner.stopPipeline(ctx, pipelineID)
+		}
+	}
+
+	return nil
+}
+
 func (runner *Runner) Run(ctx context.Context) {
 	if err := runner.synchronizePipelines(ctx); err != nil {
 		runner.logger.Errorf("starting pipelines: %s", err)
@@ -228,13 +282,7 @@ func (runner *Runner) GetPipeline(id string) *PipelineHandler {
 	runner.mu.Lock()
 	defer runner.mu.Unlock()
 
-	for p, pipeline := range runner.pipelines {
-		if id == p {
-			return pipeline
-		}
-	}
-
-	return nil
+	return runner.pipelines[id]
 }
 
 func (runner *Runner) Stop(ctx context.Context) error {

internal/replication/runner/runner_test.go

@@ -92,6 +92,10 @@ func TestRunner(t *testing.T) {
 		AnyTimes().
 		Return([]ledger.Pipeline{pipeline}, nil)
 
+	storage.EXPECT().
+		GetPipeline(gomock.Any(), pipeline.ID).
+		Return(&pipeline, nil)
+
 	storage.EXPECT().
 		OpenLedger(gomock.Any(), pipelineConfiguration.Ledger).
 		Return(logFetcher, &ledger.Ledger{}, nil)
@@ -101,7 +105,7 @@ func TestRunner(t *testing.T) {
 		Return(nil)
 
 	runner := startRunner(t, ctx, storage, connectorFactory)
-	_, err := runner.StartPipeline(ctx, pipeline)
+	_, err := runner.StartPipeline(ctx, pipeline.ID)
 	require.NoError(t, err)
 
 	connector.EXPECT().
@@ -110,7 +114,7 @@ func TestRunner(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		return runner.GetConnector("connector") != nil
-	}, time.Second, 10*time.Millisecond)
+	}, 5*time.Second, 10*time.Millisecond)
 
 	select {
 	case <-runner.GetConnector("connector").Ready():

internal/replication/runner/store.go

@@ -27,13 +27,18 @@ type Storage interface {
 	OpenLedger(context.Context, string) (LogFetcher, *ledger.Ledger, error)
 	StorePipelineState(ctx context.Context, id string, lastLogID int) error
 	ListEnabledPipelines(ctx context.Context) ([]ledger.Pipeline, error)
+	GetPipeline(ctx context.Context, id string) (*ledger.Pipeline, error)
 }
 
 type storageAdapter struct {
 	storageDriver *driver.Driver
 	systemStore   *systemstore.DefaultStore
 }
 
+func (s *storageAdapter) GetPipeline(ctx context.Context, id string) (*ledger.Pipeline, error) {
+	return s.systemStore.GetPipeline(ctx, id)
+}
+
 func (s *storageAdapter) OpenLedger(ctx context.Context, name string) (LogFetcher, *ledger.Ledger, error) {
 	store, l, err := s.storageDriver.OpenLedger(ctx, name)
 

internal/replication/runner/store_generated_test.go

@@ -248,6 +248,7 @@ func GetMigrator(db *bun.DB, options ...migrations.Option) *migrations.Migrator
 					    enabled bool,
 					    last_log_id bigint,
 					    error varchar,
+					    version int,
 					    
 					    primary key(id)
 					);

internal/storage/system/migrations.go

@@ -210,11 +210,16 @@ func (store *DefaultStore) CreatePipeline(ctx context.Context, pipeline ledger.P
 }
 
 func (store *DefaultStore) UpdatePipeline(ctx context.Context, id string, o map[string]any) error {
-	_, err := store.db.NewUpdate().
-		Table("_system.pipelines").
-		Model(&o).
-		Where("id = ?", id).
-		Exec(ctx)
+	updateQuery := store.db.NewUpdate().
+		Table("_system.pipelines")
+	for k, v := range o {
+		updateQuery = updateQuery.Set(k + " = ?", v)
+	}
+	updateQuery = updateQuery.
+		Set("version = version + 1").
+		Where("id = ?", id)
+
+	_, err := updateQuery.Exec(ctx)
 	return postgres.ResolveError(err)
 }
 

internal/storage/system/store.go

@@ -270,6 +270,7 @@ func TestUpdatePipeline(t *testing.T) {
 	require.False(t, pipelineFromDB.Enabled)
 
 	pipelineFromDB.Enabled = true
+	pipelineFromDB.Version -= 1
 	require.Equal(t, alivePipeline, *pipelineFromDB)
 }