fix: missing inserted at in logs

Author: gfyrag

go.mod

@@ -125,7 +125,7 @@ require (
 	github.com/buger/jsonparser v1.1.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/containerd/continuity v0.4.5 // indirect
-	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
 	github.com/dlclark/regexp2 v1.11.5 // indirect
 	github.com/dnwe/otelsarama v0.0.0-20240308230250-9388d9d40bc0 // indirect
 	github.com/docker/cli v27.4.1+incompatible // indirect

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

@@ -2,8 +2,11 @@ package replication
 
 import (
 	"context"
+	"fmt"
+	"runtime/debug"
 	"time"
 
+	"github.com/davecgh/go-spew/spew"
 	"github.com/formancehq/go-libs/v3/bun/bunpaginate"
 	"github.com/formancehq/go-libs/v3/collectionutils"
 	"github.com/formancehq/go-libs/v3/logging"
@@ -64,17 +67,31 @@ type PipelineHandler struct {
 }
 
 func (p *PipelineHandler) Run(ctx context.Context, ingestedLogs chan uint64) {
+	p.logger.Debugf("Pipeline started.")
 	nextInterval := time.Duration(0)
+
+	defer func() {
+		fmt.Println("Terminated.")
+		if e := recover(); e != nil {
+			fmt.Printf("error: %s\r\n", e)
+			debug.PrintStack()
+		}
+	}()
+
 	for {
 		select {
 		case ch := <-p.stopChannel:
+			p.logger.Debugf("Pipeline terminated.")
 			close(ch)
 			return
 		case <-time.After(nextInterval):
+			p.logger.Debugf("Fetch next batch.")
+			spew.Dump(p.pipeline)
 			var builder query.Builder
 			if p.pipeline.LastLogID != nil {
 				builder = query.Gt("id", *p.pipeline.LastLogID)
 			}
+			spew.Dump(builder)
 			logs, err := p.store.ListLogs(ctx, common.InitialPaginatedQuery[any]{
 				PageSize: p.pipelineConfig.LogsPageSize,
 				Column:   "id",
@@ -93,18 +110,21 @@ func (p *PipelineHandler) Run(ctx context.Context, ingestedLogs chan uint64) {
 				}
 			}
 
+			p.logger.Debugf("Got %d items", len(logs.Data))
 			if len(logs.Data) == 0 {
 				nextInterval = p.pipelineConfig.PullInterval
 				continue
 			}
 
 			for {
+				p.logger.Debugf("send to exporter")
 				_, err := p.exporter.Accept(ctx, collectionutils.Map(logs.Data, func(log ledger.Log) drivers.LogWithLedger {
 					return drivers.LogWithLedger{
 						Log:    log,
 						Ledger: p.pipeline.Ledger,
 					}
 				})...)
+				p.logger.Debugf("exported: %s", err)
 				if err != nil {
 					p.logger.Errorf("Error pushing data on exporter: %s, waiting for: %s", err, p.pipelineConfig.PushRetryPeriod)
 					select {
@@ -118,6 +138,7 @@ func (p *PipelineHandler) Run(ctx context.Context, ingestedLogs chan uint64) {
 			}
 
 			lastLogID := logs.Data[len(logs.Data)-1].ID
+			p.logger.Debugf("Move last log id to %d", lastLogID)
 			p.pipeline.LastLogID = lastLogID
 
 			select {

internal/replication/store.go

@@ -60,7 +60,12 @@ func (s *storageAdapter) OpenLedger(ctx context.Context, name string) (LogFetche
 	store, l, err := s.storageDriver.OpenLedger(ctx, name)
 
 	return LogFetcherFn(func(ctx context.Context, query common.PaginatedQuery[any]) (*bunpaginate.Cursor[ledger.Log], error) {
-		return store.Logs().Paginate(ctx, query)
+		ret, err := store.Logs().Paginate(ctx, query)
+		if err != nil {
+			return nil, err
+		}
+
+		return ret, nil
 	}), l, err
 }
 

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/42-fix-missing-inserted-at-in-log-data/notes.yaml

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

internal/storage/bucket/migrations/42-fix-missing-inserted-at-in-log-data/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';
+		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
+$$;
+

tools/provisioner/Dockerfile

@@ -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

tools/provisioner/justfile

@@ -2,9 +2,9 @@
 
 set positional-arguments
 
-push-image version='latest':
+push-image version='latest' registry='ghcr.io':
     docker buildx build . \
         --build-context root=../.. \
-        -t ghcr.io/formancehq/ledger-provisioner:{{ version }} \
+        -t {{ registry }}/formancehq/ledger-provisioner:{{ version }} \
         --push \
         --platform linux/amd64,linux/arm64