diff --git a/internal/log/log.go b/internal/log/log.go index 56a80cc6..a801347d 100644 --- a/internal/log/log.go +++ b/internal/log/log.go @@ -3,6 +3,7 @@ package log import ( "context" "os" + "time" "github.com/cybertec-postgresql/pg_timetable/internal/config" "github.com/jackc/pgx/v5/tracelog" @@ -119,6 +120,15 @@ func WithLogger(ctx context.Context, logger LoggerIface) context.Context { return context.WithValue(ctx, loggerKey{}, logger) } +// DetachedTimeout bounds cleanup work that must run even when the caller's context is cancelled. +var DetachedTimeout = 30 * time.Second // var so tests can shorten it + +// DetachedContext returns a context that survives cancellation of ctx but keeps its logger, +// bounded by DetachedTimeout so a dead database cannot block the caller forever. +func DetachedContext(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(ctx), DetachedTimeout) +} + // FallbackLogger is an alias for the standard logger var FallbackLogger = logrus.StandardLogger() diff --git a/internal/pgengine/access.go b/internal/pgengine/access.go index b9e93124..05aaa580 100644 --- a/internal/pgengine/access.go +++ b/internal/pgengine/access.go @@ -66,8 +66,13 @@ SELECT $1, $2 WHERE return res.RowsAffected() == 1 } +// RemoveChainRunStatus frees one run's slot, so other running instances keep theirs func (pge *PgEngine) RemoveChainRunStatus(ctx context.Context, chainID int) { - const sqlRemoveRunStatus = `DELETE FROM timetable.active_chain WHERE chain_id = $1 and client_name = $2` + const sqlRemoveRunStatus = `DELETE FROM timetable.active_chain WHERE ctid = ( + SELECT ctid FROM timetable.active_chain + WHERE chain_id = $1 AND client_name = $2 + LIMIT 1 FOR UPDATE SKIP LOCKED +)` _, err := pge.ConfigDb.Exec(ctx, sqlRemoveRunStatus, chainID, pge.ClientName) if err != nil { pge.l.WithError(err).Error("Cannot save information about the chain run status") diff --git a/internal/pgengine/pgengine_test.go b/internal/pgengine/pgengine_test.go index e9a68d06..0516f5a4 100644 --- a/internal/pgengine/pgengine_test.go +++ b/internal/pgengine/pgengine_test.go @@ -109,7 +109,7 @@ func TestSchedulerFunctions(t *testing.T) { assert.Greater(t, txid, int64(0), "Should return transaction id") assert.NoError(t, pge.GetChainElements(ctx, &chains, 0), "Should no error in clean database") assert.Empty(t, chains, "Should be empty in clean database") - pge.CommitTransaction(ctx, tx) + assert.NoError(t, tx.Commit(ctx)) }) t.Run("Check GetChainElements returns only live tasks", func(t *testing.T) { @@ -137,7 +137,7 @@ func TestSchedulerFunctions(t *testing.T) { TaskID: 0, ChainID: 0}), "Should no error in clean database") assert.Empty(t, paramVals, "Should be empty in clean database") - pge.CommitTransaction(ctx, tx) + assert.NoError(t, tx.Commit(ctx)) }) t.Run("Check InsertChainRunStatus function", func(t *testing.T) { @@ -147,6 +147,18 @@ func TestSchedulerFunctions(t *testing.T) { assert.True(t, res, "Active chain should be inserted") }) + t.Run("Check RemoveChainRunStatus removes one instance only", func(t *testing.T) { + const chainID = 4242 + require.True(t, pge.InsertChainRunStatus(ctx, chainID, 16)) + require.True(t, pge.InsertChainRunStatus(ctx, chainID, 16)) + pge.RemoveChainRunStatus(ctx, chainID) + var count int + require.NoError(t, pge.ConfigDb.QueryRow(ctx, + "SELECT count(*) FROM timetable.active_chain WHERE chain_id = $1", chainID).Scan(&count)) + assert.Equal(t, 1, count, "Other running instance must keep its slot") + pge.RemoveChainRunStatus(ctx, chainID) + }) + t.Run("Check ExecuteSQLCommand function", func(t *testing.T) { tx, txid, err := pge.StartTransaction(ctx) assert.NoError(t, err, "Should start transaction") @@ -163,7 +175,7 @@ func TestSchedulerFunctions(t *testing.T) { assert.NoError(t, f("SELECT $1::int4", []string{"[42]", `[14]`}), "Simple query with doubled parameters") assert.NoError(t, f("SELECT $1::int4, $2::text", []string{`[42, "hey"]`}), "Simple query with two parameters") - pge.CommitTransaction(ctx, tx) + assert.NoError(t, tx.Commit(ctx)) }) } diff --git a/internal/pgengine/transaction.go b/internal/pgengine/transaction.go index 176c87e9..6817dbb2 100644 --- a/internal/pgengine/transaction.go +++ b/internal/pgengine/transaction.go @@ -16,21 +16,19 @@ import ( // StartTransaction returns transaction object, virtual transaction id and error func (pge *PgEngine) StartTransaction(ctx context.Context) (tx pgx.Tx, vxid int64, err error) { if tx, err = pge.ConfigDb.Begin(ctx); err != nil { - return + return nil, 0, err } err = tx.QueryRow(ctx, `SELECT (split_part(virtualxid, '/', 1)::int8 << 32) | split_part(virtualxid, '/', 2)::int8 FROM pg_locks WHERE pid = pg_backend_pid() AND virtualxid IS NOT NULL`).Scan(&vxid) - return -} - -// CommitTransaction commits transaction and log error in the case of error -func (pge *PgEngine) CommitTransaction(ctx context.Context, tx pgx.Tx) { - err := tx.Commit(ctx) if err != nil { - log.GetLogger(ctx).WithError(err).Error("Application cannot commit after job finished") + bctx, cancel := log.DetachedContext(ctx) + defer cancel() + pge.RollbackTransaction(bctx, tx) + return nil, 0, err } + return tx, vxid, nil } // RollbackTransaction rollbacks transaction and log error in the case of error diff --git a/internal/pgengine/transaction_test.go b/internal/pgengine/transaction_test.go index 5c495fed..75549b5d 100644 --- a/internal/pgengine/transaction_test.go +++ b/internal/pgengine/transaction_test.go @@ -44,8 +44,17 @@ func TestStartTransaction(t *testing.T) { assert.Error(t, err) mockPool.ExpectBegin() - mockPool.ExpectQuery("SELECT").WillReturnRows(pgxmock.NewRows([]string{"txid"}).AddRow(int64(42))) + mockPool.ExpectQuery("SELECT").WillReturnError(errors.New("foo")) + mockPool.ExpectRollback() tx, txid, err := pge.StartTransaction(ctx) + assert.Nil(t, tx, "half-started transaction must be rolled back, not returned") + assert.Zero(t, txid) + assert.Error(t, err) + assert.NoError(t, mockPool.ExpectationsWereMet()) + + mockPool.ExpectBegin() + mockPool.ExpectQuery("SELECT").WillReturnRows(pgxmock.NewRows([]string{"txid"}).AddRow(int64(42))) + tx, txid, err = pge.StartTransaction(ctx) assert.NotNil(t, tx) assert.EqualValues(t, 42, txid) assert.NoError(t, err) @@ -56,15 +65,9 @@ func TestMustTransaction(t *testing.T) { pge := pgengine.NewDB(mockPool, "pgengine_unit_test") - mockPool.ExpectBegin() - mockPool.ExpectCommit().WillReturnError(errors.New("error")) - tx, err := mockPool.Begin(context.Background()) - assert.NoError(t, err) - pge.CommitTransaction(ctx, tx) - mockPool.ExpectBegin() mockPool.ExpectRollback().WillReturnError(errors.New("error")) - tx, err = mockPool.Begin(context.Background()) + tx, err := mockPool.Begin(context.Background()) assert.NoError(t, err) pge.RollbackTransaction(ctx, tx) diff --git a/internal/scheduler/chain.go b/internal/scheduler/chain.go index 559df7b0..7291731b 100644 --- a/internal/scheduler/chain.go +++ b/internal/scheduler/chain.go @@ -159,6 +159,7 @@ func (sch *Scheduler) chainWorker(ctx context.Context, chains <-chan Chain) { chainContext, cancel := context.WithCancel(chainContext) sch.addActiveChain(chain.ChainID, cancel) sch.executeChain(chainContext, chain) + sch.removeChainRunStatus(chainContext, chain.ChainID) sch.deleteActiveChain(chain.ChainID) cancel() sch.Unlock(chain.ExclusiveExecution) @@ -178,6 +179,14 @@ func getTimeoutContext(ctx context.Context, globalTimeout int, customTimeout int return ctx, nil } +// removeChainRunStatus frees the chain's active_chain slot. Runs detached so it +// survives a cancelled scheduler context and cannot hang on a dead database. +func (sch *Scheduler) removeChainRunStatus(ctx context.Context, chainID int) { + bctx, cancel := log.DetachedContext(ctx) + defer cancel() + sch.pgengine.RemoveChainRunStatus(bctx, chainID) +} + func (sch *Scheduler) executeOnErrorHandler(ctx context.Context, chain Chain) { if ctx.Err() != nil || chain.OnError == "" { return @@ -191,9 +200,8 @@ func (sch *Scheduler) executeOnErrorHandler(ctx context.Context, chain Chain) { } sch.pgengine.SetCurrentTaskContext(ctx, tx, chain.ChainID, 0) _, err = tx.Exec(ctx, chain.OnError) - sch.pgengine.CommitTransaction(ctx, tx) - if err != nil { - l.Info("Error handler failed") + if err = cmp.Or(err, tx.Commit(ctx)); err != nil { + l.WithError(err).Info("Error handler failed") return } l.Info("Error handler executed successfully") @@ -202,8 +210,6 @@ func (sch *Scheduler) executeOnErrorHandler(ctx context.Context, chain Chain) { /* execute a chain of tasks */ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { var ChainTasks []pgengine.ChainTask - var bctx context.Context - var cancel context.CancelFunc var vxid int64 // OTel tracing: create root span for this chain execution @@ -224,17 +230,29 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { } chainL := sch.l.WithField("chain", chain) + chainFailed := func(l log.LoggerIface, tx pgx.Tx, err error, msg string) { + l.WithError(err).Error(msg) + chainSpan.SetStatus(codes.Error, msg) + lctx := log.WithLogger(ctx, l) + if tx != nil { + bctx, cancel := log.DetachedContext(lctx) + sch.pgengine.RollbackTransaction(bctx, tx) + cancel() + } + sch.provider.RecordChainFailed(context.WithoutCancel(lctx), sch.Config().ClientName) + sch.executeOnErrorHandler(lctx, chain) // skipped if the chain was stopped + } + tx, vxid, err := sch.pgengine.StartTransaction(chainCtx) if err != nil { - chainL.WithError(err).Error("Cannot start transaction") + chainFailed(chainL, nil, err, "Chain failed to start") return } chainL = chainL.WithField("vxid", vxid) err = sch.pgengine.GetChainElements(chainCtx, &ChainTasks, chain.ChainID) if err != nil { - chainL.WithError(err).Error("Failed to retrieve chain elements") - sch.pgengine.RollbackTransaction(chainCtx, tx) + chainFailed(chainL, tx, err, "Chain failed to load tasks") return } @@ -252,30 +270,25 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { l.Info("Task executed successfully") } - // we detach the context from cancellation here because the current one - // (chainCtx and its parent ctx) might be cancelled, e.g. by notify_chain_stop(). - // Cleanup operations below must still run to keep timetable.active_chain consistent. - bctx = log.WithLogger(context.WithoutCancel(ctx), l) if err != nil { if !task.IgnoreError { - chainL.Error("Chain failed") - sch.pgengine.RemoveChainRunStatus(bctx, chain.ChainID) - sch.pgengine.RollbackTransaction(bctx, tx) - chainSpan.SetStatus(codes.Error, "chain failed") - sch.provider.RecordChainFailed(bctx, sch.Config().ClientName) - sch.executeOnErrorHandler(bctx, chain) + chainFailed(l, tx, err, "Chain failed") return } l.Info("Ignoring task failure") } } - bctx = log.WithLogger(context.WithoutCancel(chainCtx), chainL) - sch.pgengine.CommitTransaction(bctx, tx) + // commit even if the chain was stopped meanwhile: all its tasks have succeeded + if err = tx.Commit(log.WithLogger(context.WithoutCancel(ctx), chainL)); err != nil { + chainFailed(chainL, nil, err, "Chain failed to commit") + return + } sch.provider.RecordChainDuration(ctx, time.Since(chainStart).Seconds(), sch.Config().ClientName) chainL.Info("Chain executed successfully") sch.provider.RecordChainCompleted(ctx, sch.Config().ClientName) - sch.pgengine.RemoveChainRunStatus(bctx, chain.ChainID) if chain.SelfDestruct { + bctx, cancel := log.DetachedContext(log.WithLogger(ctx, chainL)) + defer cancel() sch.pgengine.DeleteChain(bctx, chain.ChainID) } } diff --git a/internal/scheduler/chain_test.go b/internal/scheduler/chain_test.go index 100b8e85..ce811d75 100644 --- a/internal/scheduler/chain_test.go +++ b/internal/scheduler/chain_test.go @@ -77,7 +77,7 @@ func TestChainWorker(t *testing.T) { mock.ExpectBegin().WillReturnError(errors.New("expected")) mock.ExpectExec("INSERT INTO timetable\\.log").WillReturnResult(pgxmock.NewResult("INSERT", 1)) mock.ExpectExec("INSERT INTO timetable\\.log").WillReturnResult(pgxmock.NewResult("INSERT", 1)) - mock.ExpectExec("DELETE").WillReturnResult(pgxmock.NewResult("DELETE", 1)) + mock.ExpectExec("DELETE FROM timetable\\.active_chain").WillReturnResult(pgxmock.NewResult("DELETE", 1)) chains <- Chain{SelfDestruct: true} sch.chainWorker(ctx, chains) }) @@ -103,6 +103,153 @@ func TestExecuteChain(t *testing.T) { sch.executeChain(t.Context(), Chain{Timeout: 1}) } +func TestExecuteChainEarlyReturn(t *testing.T) { + c := Chain{ChainID: 42, OnError: "FOO"} + vxidRows := func() *pgxmock.Rows { return pgxmock.NewRows([]string{"vxid"}).AddRow(int64(42)) } + + tests := []struct { + name string + ctx func() context.Context + expect func(mock pgxmock.PgxPoolIface) + onError bool + }{ + { + name: "cannot start transaction", + ctx: context.Background, + expect: func(mock pgxmock.PgxPoolIface) { + mock.ExpectBegin().WillReturnError(errors.New("query_wait_timeout")) + }, + onError: true, + }, + { + name: "chain stopped before start", + ctx: func() context.Context { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + return ctx + }, + expect: func(mock pgxmock.PgxPoolIface) { + mock.ExpectBegin() + }, + }, + { + name: "cannot get virtual transaction id", + ctx: context.Background, + expect: func(mock pgxmock.PgxPoolIface) { + mock.ExpectBegin() + mock.ExpectQuery("FROM pg_locks").WillReturnError(errors.New("query_wait_timeout")) + mock.ExpectRollback() + }, + onError: true, + }, + { + name: "failed to retrieve chain elements", + ctx: context.Background, + expect: func(mock pgxmock.PgxPoolIface) { + mock.ExpectBegin() + mock.ExpectQuery("FROM pg_locks").WillReturnRows(vxidRows()) + mock.ExpectQuery("FROM timetable\\.task").WithArgs(c.ChainID).WillReturnError(errors.New("query_wait_timeout")) + mock.ExpectRollback() + }, + onError: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + mock, err := pgxmock.NewPool() + assert.NoError(t, err) + pge := pgengine.NewDB(mock, "-c", "scheduler_unit_test") + sch := New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}), otel.NewNoop()) + + tc.expect(mock) + if tc.onError { + expectOnError(mock) + } else { + mock.ExpectBegin() // must stay unconsumed: no on_error for a stopped chain + } + + sch.executeChain(tc.ctx(), c) + if tc.onError { + assert.NoError(t, mock.ExpectationsWereMet()) + } else { + assert.Error(t, mock.ExpectationsWereMet()) + } + }) + } +} + +func expectOnError(mock pgxmock.PgxPoolIface) { + mock.ExpectBegin() + mock.ExpectQuery("FROM pg_locks").WillReturnRows(pgxmock.NewRows([]string{"vxid"}).AddRow(int64(42))) + mock.ExpectExec("SELECT set_config").WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg(), pgxmock.AnyArg()).WillReturnResult(pgxmock.NewResult("SELECT", 1)) + mock.ExpectExec("FOO").WillReturnResult(pgxmock.NewResult("FOO", 1)) + mock.ExpectCommit() +} + +func TestExecuteChainCommit(t *testing.T) { + c := Chain{ChainID: 42, OnError: "FOO", SelfDestruct: true} + expectChain := func(mock pgxmock.PgxPoolIface) *pgxmock.ExpectedCommit { + mock.ExpectBegin() + mock.ExpectQuery("FROM pg_locks").WillReturnRows(pgxmock.NewRows([]string{"vxid"}).AddRow(int64(42))) + mock.ExpectQuery("FROM timetable\\.task").WithArgs(c.ChainID).WillReturnRows(pgxmock.NewRows([]string{"task_id"})) + return mock.ExpectCommit() + } + newScheduler := func(t *testing.T) (*Scheduler, pgxmock.PgxPoolIface) { + mock, err := pgxmock.NewPool() + assert.NoError(t, err) + pge := pgengine.NewDB(mock, "-c", "scheduler_unit_test") + return New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}), otel.NewNoop()), mock + } + + t.Run("commit is not bounded by DetachedTimeout", func(t *testing.T) { + defer func(d time.Duration) { log.DetachedTimeout = d }(log.DetachedTimeout) + log.DetachedTimeout = 50 * time.Millisecond + sch, mock := newScheduler(t) + expectChain(mock).WillDelayFor(150 * time.Millisecond) + mock.ExpectExec("DELETE FROM timetable\\.chain").WithArgs(c.ChainID).WillReturnResult(pgxmock.NewResult("DELETE", 1)) + sch.executeChain(t.Context(), c) + assert.NoError(t, mock.ExpectationsWereMet()) + }) + + t.Run("commit failure runs on_error and keeps chain", func(t *testing.T) { + sch, mock := newScheduler(t) + expectChain(mock).WillReturnError(errors.New("serialization failure")) + expectOnError(mock) + sch.executeChain(t.Context(), c) + assert.NoError(t, mock.ExpectationsWereMet()) + }) +} + +func TestChainWorkerRemovesRunStatusOnEarlyReturn(t *testing.T) { + const clientName = "scheduler_unit_test" + c := Chain{ChainID: 42, MaxInstances: 16} + mock, err := pgxmock.NewPool() + assert.NoError(t, err) + pge := pgengine.NewDB(mock, "-c", clientName) + sch := New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}), otel.NewNoop()) + + mock.ExpectExec("INSERT INTO timetable\\.active_chain"). + WithArgs(c.ChainID, clientName, c.MaxInstances). + WillReturnResult(pgxmock.NewResult("INSERT", 1)) + mock.ExpectBegin().WillReturnError(errors.New("query_wait_timeout")) + mock.ExpectExec("DELETE FROM timetable\\.active_chain"). + WithArgs(c.ChainID, clientName). + WillReturnResult(pgxmock.NewResult("DELETE", 1)) + + ctx, cancel := context.WithCancel(context.Background()) + chains := make(chan Chain, 1) + chains <- c + done := make(chan struct{}) + go func() { + sch.chainWorker(ctx, chains) + close(done) + }() + assert.Eventually(t, func() bool { return mock.ExpectationsWereMet() == nil }, 5*time.Second, 10*time.Millisecond) + cancel() + <-done +} + func TestExecuteChainElement(t *testing.T) { mock, err := pgxmock.NewPool() assert.NoError(t, err) @@ -132,10 +279,11 @@ func TestExecuteOnErrorHandler(t *testing.T) { assert.NoError(t, mock.ExpectationsWereMet()) }) - t.Run("check error handler if context cancelled", func(*testing.T) { + t.Run("check error handler if context cancelled", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() sch.executeOnErrorHandler(ctx, c) + assert.NoError(t, mock.ExpectationsWereMet()) }) t.Run("check error handler if begin fails", func(*testing.T) { diff --git a/internal/scheduler/interval_chain.go b/internal/scheduler/interval_chain.go index c15c27cb..1b1eded8 100644 --- a/internal/scheduler/interval_chain.go +++ b/internal/scheduler/interval_chain.go @@ -91,6 +91,7 @@ func (sch *Scheduler) intervalChainWorker(ctx context.Context, ichains <-chan In } sch.Lock(ichain.ExclusiveExecution) sch.executeChain(chainContext, ichain.Chain) + sch.removeChainRunStatus(chainContext, ichain.ChainID) sch.Unlock(ichain.ExclusiveExecution) if ichain.RepeatAfter { go sch.reschedule(chainContext, ichain) diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 3f8309a7..8cf4f508 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -33,17 +33,16 @@ func TestExecuteChainCancelledContextCleansUp(t *testing.T) { VALUES ($1, 1, 'SQL', 'SELECT pg_sleep(0.5)')`, chainID) require.NoError(t, err) - // Simulate what chainWorker does before calling executeChain. - require.True(t, pge.InsertChainRunStatus(ctx, chainID, 16), "chain run status should be inserted") - - // Cancel the context while the task is sleeping, mimicking notify_chain_stop(). + // Cancel the worker context while the task is sleeping, mimicking notify_chain_stop(). execCtx, cancel := context.WithCancel(ctx) go func() { time.Sleep(100 * time.Millisecond) cancel() }() - sch.executeChain(execCtx, Chain{ChainID: chainID, MaxInstances: 16}) + chains := make(chan Chain, 1) + chains <- Chain{ChainID: chainID, MaxInstances: 16} + sch.chainWorker(execCtx, chains) // The stale active_chain row must be gone: cleanup must succeed despite the // cancelled context. @@ -54,6 +53,41 @@ func TestExecuteChainCancelledContextCleansUp(t *testing.T) { assert.Zero(t, count, "active_chain entry must be removed even when the execution context was cancelled") } +func TestExecuteChainCommitsAfterDetachedTimeout(t *testing.T) { + container, cleanup := testutils.SetupPostgresContainer(t) + defer cleanup() + + // A chain that runs longer than DetachedTimeout must still commit: + // the timeout bounds cleanup, not the chain. + defer func(d time.Duration) { log.DetachedTimeout = d }(log.DetachedTimeout) + log.DetachedTimeout = 200 * time.Millisecond + + pge := container.Engine + ctx := context.Background() + sch := New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}), otel.NewNoop()) + + _, err := pge.ConfigDb.Exec(ctx, "CREATE TABLE detached_timeout_test (id int)") + require.NoError(t, err) + + var chainID int + err = pge.ConfigDb.QueryRow(ctx, ` + INSERT INTO timetable.chain(chain_name, max_instances) + VALUES ('detached_timeout_test', 16) RETURNING chain_id`).Scan(&chainID) + require.NoError(t, err) + + _, err = pge.ConfigDb.Exec(ctx, ` + INSERT INTO timetable.task(chain_id, task_order, kind, command, autonomous) + VALUES ($1, 1, 'SQL', 'INSERT INTO detached_timeout_test SELECT 1 FROM pg_sleep(0.5)', false)`, chainID) + require.NoError(t, err) + + sch.executeChain(ctx, Chain{ChainID: chainID, MaxInstances: 16}) + + var count int + err = pge.ConfigDb.QueryRow(ctx, "SELECT count(*) FROM detached_timeout_test").Scan(&count) + require.NoError(t, err) + assert.Equal(t, 1, count, "chain transaction must be committed") +} + func TestRun(t *testing.T) { container, cleanup := testutils.SetupPostgresContainer(t) defer cleanup()