fix: missing inserted at in logs

Also fixes stop issue of exporters.

Author: gfyrag

internal/replication/drivers/clickhouse/driver.go

@@ -44,6 +44,8 @@ func (c *Driver) Start(ctx context.Context) error {
 
 func (c *Driver) Accept(ctx context.Context, logs ...drivers.LogWithLedger) ([]error, error) {
 
+	c.logger.Debugf("Prepare new batch of %d logs", len(logs))
+
 	batch, err := c.db.PrepareBatch(ctx, "insert into logs(ledger, id, type, date, data)")
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to prepare batch")

internal/replication/manager.go

@@ -217,7 +217,12 @@ func (m *Manager) synchronizePipelines(ctx context.Context) error {
 	for _, pipeline := range pipelines {
 		m.logger.Debugf("restoring pipeline %s", pipeline.ID)
 		if _, err := m.startPipeline(ctx, pipeline); err != nil {
-			return err
+			switch {
+			case errors.Is(err, ledger.ErrAlreadyStarted("")):
+				m.logger.Debugf("Pipeline already started, skipping")
+			default:
+				return err
+			}
 		}
 	}
 

internal/replication/pipeline.go

@@ -64,13 +64,21 @@ type PipelineHandler struct {
 }
 
 func (p *PipelineHandler) Run(ctx context.Context, ingestedLogs chan uint64) {
+	p.logger.Debugf("Pipeline started.")
 	nextInterval := time.Duration(0)
+
+	stop := func(ch chan error) {
+		p.logger.Debugf("Pipeline terminated.")
+		close(ch)
+	}
+
 	for {
 		select {
 		case ch := <-p.stopChannel:
-			close(ch)
+			stop(ch)
 			return
 		case <-time.After(nextInterval):
+			p.logger.Debugf("Fetch next batch.")
 			var builder query.Builder
 			if p.pipeline.LastLogID != nil {
 				builder = query.Gt("id", *p.pipeline.LastLogID)
@@ -86,38 +94,57 @@ func (p *PipelineHandler) Run(ctx context.Context, ingestedLogs chan uint64) {
 			if err != nil {
 				p.logger.Errorf("Error fetching logs: %s", err)
 				select {
-				case <-ctx.Done():
+				case ch := <-p.stopChannel:
+					stop(ch)
 					return
 				case <-time.After(p.pipelineConfig.PullInterval):
 					continue
 				}
 			}
 
+			p.logger.Debugf("Got %d items", len(logs.Data))
 			if len(logs.Data) == 0 {
 				nextInterval = p.pipelineConfig.PullInterval
 				continue
 			}
 
 			for {
-				_, err := p.exporter.Accept(ctx, collectionutils.Map(logs.Data, func(log ledger.Log) drivers.LogWithLedger {
-					return drivers.LogWithLedger{
-						Log:    log,
-						Ledger: p.pipeline.Ledger,
-					}
-				})...)
-				if err != nil {
-					p.logger.Errorf("Error pushing data on exporter: %s, waiting for: %s", err, p.pipelineConfig.PushRetryPeriod)
-					select {
-					case <-ctx.Done():
-						return
-					case <-time.After(p.pipelineConfig.PushRetryPeriod):
-						continue
+				p.logger.Debugf("Send data to exporter.")
+				errChan := make(chan error, 1)
+				exportContext, cancel := context.WithCancel(ctx)
+				go func() {
+					_, err := p.exporter.Accept(exportContext, collectionutils.Map(logs.Data, func(log ledger.Log) drivers.LogWithLedger {
+						return drivers.LogWithLedger{
+							Log:    log,
+							Ledger: p.pipeline.Ledger,
+						}
+					})...)
+					errChan <- err
+				}()
+				select {
+				case err := <-errChan:
+					cancel()
+					if err != nil {
+						p.logger.Errorf("Error pushing data on exporter: %s, waiting for: %s", err, p.pipelineConfig.PushRetryPeriod)
+						select {
+						case ch := <-p.stopChannel:
+							stop(ch)
+							return
+						case <-time.After(p.pipelineConfig.PushRetryPeriod):
+							continue
+						}
 					}
+				case ch := <-p.stopChannel:
+					cancel()
+					stop(ch)
+					return
 				}
+
 				break
 			}
 
 			lastLogID := logs.Data[len(logs.Data)-1].ID
+			p.logger.Debugf("Move last log id to %d", *lastLogID)
 			p.pipeline.LastLogID = lastLogID
 
 			select {
@@ -129,14 +156,15 @@ func (p *PipelineHandler) Run(ctx context.Context, ingestedLogs chan uint64) {
 			if !logs.HasMore {
 				nextInterval = p.pipelineConfig.PullInterval
 			} else {
+				p.logger.Debugf("Has more logs to fetch.")
 				nextInterval = 0
 			}
 		}
 	}
 }
 
 func (p *PipelineHandler) Shutdown(ctx context.Context) error {
-	p.logger.Infof("shutdowning pipeline")
+	p.logger.Infof("Shutdowning pipeline")
 	errorChannel := make(chan error, 1)
 	select {
 	case <-ctx.Done():

internal/storage/bucket/migrations/41-add-missing-index/notes.yaml

@@ -0,0 +1 @@
+name: Add missing index

internal/storage/bucket/migrations/41-add-missing-index/up.sql

@@ -0,0 +1 @@
+create index {{ if not .Transactional }}concurrently{{end}} logs_ids on "{{.Schema}}".logs (id);

internal/storage/bucket/migrations/41-remove-useless-trigger/notes.yaml renamed to internal/storage/bucket/migrations/42-remove-useless-trigger/notes.yaml

@@ -0,0 +1 @@
+name: Fill inserted at field in log data

internal/storage/bucket/migrations/41-remove-useless-trigger/up.sql renamed to internal/storage/bucket/migrations/42-remove-useless-trigger/up.sql

@@ -0,0 +1,40 @@
+do $$
+	declare
+		_offset integer := 0;
+		_batch_size integer := 1000;
+	begin
+		set search_path = '{{ .Schema }}';
+
+		create temp table logs_view as
+			select row_number() over (order by id) as row_number, id, ledger
+			from logs
+			where type = 'NEW_TRANSACTION' or type = 'REVERTED_TRANSACTION';
+		create index logs_view_row_numbers on logs_view(row_number);
+
+		perform pg_notify('migrations-{{ .Schema }}', 'init: ' || (select count(*) from logs_view));
+
+		loop
+			with _rows as (
+				select id, ledger, row_number
+				from logs_view
+				where row_number >= _offset and row_number < _offset + _batch_size
+			)
+			update logs
+			set data = jsonb_set(data, '{transaction, insertedAt}', to_jsonb(to_jsonb(date)#>>'{}' || 'Z'))
+			from _rows
+			where logs.id = _rows.id and
+			      logs.ledger = _rows.ledger;
+
+			exit when not found;
+
+			_offset = _offset + _batch_size;
+
+			perform pg_notify('migrations-{{ .Schema }}', 'continue: ' || _batch_size);
+
+			commit;
+		end loop;
+
+		drop table if exists logs_view;
+	end
+$$;
+

internal/storage/bucket/migrations/43-fix-missing-inserted-at-in-log-data/notes.yaml

@@ -1,3 +1,4 @@
+# syntax=docker/dockerfile:1
 FROM golang:1.25-alpine AS compiler
 WORKDIR /src
 COPY --from=root pkg pkg
@@ -14,6 +15,7 @@ COPY cmd /src/tools/provisioner/cmd
 COPY pkg /src/tools/provisioner/pkg
 RUN --mount=type=cache,target=$GOPATH go build -o provisioner
 
+# syntax=docker/dockerfile:1
 FROM alpine:3.21
 LABEL org.opencontainers.image.source=https://github.com/formancehq/ledger
 COPY --from=compiler /src/tools/provisioner/provisioner /bin/provisioner