From 78dfd3c8c33deab741b87922564992d39aacedb3 Mon Sep 17 00:00:00 2001 From: olaberglund Date: Thu, 8 Oct 2026 09:50:54 +0200 Subject: [PATCH 1/6] [-] fix `active_chain` row leak when chain cannot start or load tasks `executeChain()` returned early on `StartTransaction()` and `GetChainElements()` errors without calling `RemoveChainRunStatus()`, so the row inserted by the worker stayed in `timetable.active_chain`. Once the leaked rows reached `max_instances` the chain was never started again. Remove the run status on both paths using a context detached from cancellation, as the task failure path already does. --- internal/scheduler/chain.go | 6 +++- internal/scheduler/chain_test.go | 57 ++++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) diff --git a/internal/scheduler/chain.go b/internal/scheduler/chain.go index 559df7b0..ae934fc5 100644 --- a/internal/scheduler/chain.go +++ b/internal/scheduler/chain.go @@ -227,6 +227,8 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { tx, vxid, err := sch.pgengine.StartTransaction(chainCtx) if err != nil { chainL.WithError(err).Error("Cannot start transaction") + bctx = log.WithLogger(context.WithoutCancel(ctx), chainL) + sch.pgengine.RemoveChainRunStatus(bctx, chain.ChainID) return } chainL = chainL.WithField("vxid", vxid) @@ -234,7 +236,9 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { 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) + bctx = log.WithLogger(context.WithoutCancel(ctx), chainL) + sch.pgengine.RollbackTransaction(bctx, tx) + sch.pgengine.RemoveChainRunStatus(bctx, chain.ChainID) return } diff --git a/internal/scheduler/chain_test.go b/internal/scheduler/chain_test.go index 100b8e85..ffc0a8f3 100644 --- a/internal/scheduler/chain_test.go +++ b/internal/scheduler/chain_test.go @@ -103,6 +103,63 @@ func TestExecuteChain(t *testing.T) { sch.executeChain(t.Context(), Chain{Timeout: 1}) } +func TestExecuteChainRemovesRunStatusOnEarlyReturn(t *testing.T) { + const clientName = "scheduler_unit_test" + c := Chain{ChainID: 42, MaxInstances: 16} + + tests := []struct { + name string + ctx func() context.Context + expect func(mock pgxmock.PgxPoolIface) + }{ + { + name: "cannot start transaction", + ctx: context.Background, + expect: func(mock pgxmock.PgxPoolIface) { + mock.ExpectBegin().WillReturnError(errors.New("query_wait_timeout")) + }, + }, + { + name: "cannot start transaction with cancelled context", + ctx: func() context.Context { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + return ctx + }, + expect: func(mock pgxmock.PgxPoolIface) { + mock.ExpectBegin() + }, + }, + { + name: "failed to retrieve chain elements", + ctx: context.Background, + expect: func(mock pgxmock.PgxPoolIface) { + mock.ExpectBegin() + mock.ExpectQuery("SELECT").WillReturnRows(pgxmock.NewRows([]string{"vxid"}).AddRow(int64(42))) + mock.ExpectQuery("FROM timetable\\.task").WithArgs(c.ChainID).WillReturnError(errors.New("query_wait_timeout")) + mock.ExpectRollback() + }, + }, + } + + 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", clientName) + sch := New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}), otel.NewNoop()) + + tc.expect(mock) + mock.ExpectExec("DELETE FROM timetable\\.active_chain"). + WithArgs(c.ChainID, clientName). + WillReturnResult(pgxmock.NewResult("DELETE", 1)) + + sch.executeChain(tc.ctx(), c) + assert.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + func TestExecuteChainElement(t *testing.T) { mock, err := pgxmock.NewPool() assert.NoError(t, err) From 974381b539c2acf7bde1017547bc003285d30681 Mon Sep 17 00:00:00 2001 From: Pavlo Golub Date: Thu, 8 Oct 2026 17:55:07 +0200 Subject: [PATCH 2/6] [-] remove `active_chain` row in the worker so no `executeChain` exit can leak it Both workers now call `removeChainRunStatus` right after `executeChain` returns, pairing the delete with the insert in one place. Cleanup runs on a detached context bounded by `log.DetachedTimeout` so a cancelled scheduler context cannot skip it and a dead database cannot block a worker forever. Chains that fail before any task runs now count as failed: span status, `RecordChainFailed` and the `on_error` handler fire on every failure path. `StartTransaction` rolls back its own transaction when the vxid lookup fails instead of handing a live `tx` back with an error. --- internal/log/log.go | 10 +++++ internal/pgengine/transaction.go | 10 ++++- internal/pgengine/transaction_test.go | 11 ++++- internal/scheduler/chain.go | 45 +++++++++++-------- internal/scheduler/chain_test.go | 62 ++++++++++++++++++++++----- internal/scheduler/interval_chain.go | 1 + internal/scheduler/scheduler_test.go | 9 ++-- 7 files changed, 111 insertions(+), 37 deletions(-) diff --git a/internal/log/log.go b/internal/log/log.go index 56a80cc6..eb1b3c95 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. +const DetachedTimeout = 30 * time.Second + +// 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/transaction.go b/internal/pgengine/transaction.go index 176c87e9..31ca4c20 100644 --- a/internal/pgengine/transaction.go +++ b/internal/pgengine/transaction.go @@ -16,13 +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 + if err != nil { + bctx, cancel := log.DetachedContext(ctx) + defer cancel() + pge.RollbackTransaction(bctx, tx) + return nil, 0, err + } + return tx, vxid, nil } // CommitTransaction commits 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..d8fa9eb5 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) diff --git a/internal/scheduler/chain.go b/internal/scheduler/chain.go index ae934fc5..2a58611e 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,8 +179,16 @@ 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 == "" { + if chain.OnError == "" { return } l := sch.l.WithField("chain", chain) @@ -202,8 +211,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,21 +231,31 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { } chainL := sch.l.WithField("chain", chain) + // cleanup (rollback, commit, on_error) must run even if ctx is cancelled, e.g. by notify_chain_stop() + bctx, bcancel := log.DetachedContext(log.WithLogger(ctx, chainL)) + defer bcancel() + + chainFailed := func(msg string) { + chainL.Error(msg) + chainSpan.SetStatus(codes.Error, msg) + sch.provider.RecordChainFailed(bctx, sch.Config().ClientName) + sch.executeOnErrorHandler(bctx, chain) + } + tx, vxid, err := sch.pgengine.StartTransaction(chainCtx) if err != nil { chainL.WithError(err).Error("Cannot start transaction") - bctx = log.WithLogger(context.WithoutCancel(ctx), chainL) - sch.pgengine.RemoveChainRunStatus(bctx, chain.ChainID) + chainFailed("Chain failed to start") return } chainL = chainL.WithField("vxid", vxid) + bctx = log.WithLogger(bctx, chainL) err = sch.pgengine.GetChainElements(chainCtx, &ChainTasks, chain.ChainID) if err != nil { chainL.WithError(err).Error("Failed to retrieve chain elements") - bctx = log.WithLogger(context.WithoutCancel(ctx), chainL) sch.pgengine.RollbackTransaction(bctx, tx) - sch.pgengine.RemoveChainRunStatus(bctx, chain.ChainID) + chainFailed("Chain failed to load tasks") return } @@ -256,29 +273,19 @@ 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) + sch.pgengine.RollbackTransaction(log.WithLogger(bctx, l), tx) + chainFailed("Chain failed") return } l.Info("Ignoring task failure") } } - bctx = log.WithLogger(context.WithoutCancel(chainCtx), chainL) sch.pgengine.CommitTransaction(bctx, tx) 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 { sch.pgengine.DeleteChain(bctx, chain.ChainID) } diff --git a/internal/scheduler/chain_test.go b/internal/scheduler/chain_test.go index ffc0a8f3..4caa5152 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,9 +103,9 @@ func TestExecuteChain(t *testing.T) { sch.executeChain(t.Context(), Chain{Timeout: 1}) } -func TestExecuteChainRemovesRunStatusOnEarlyReturn(t *testing.T) { - const clientName = "scheduler_unit_test" - c := Chain{ChainID: 42, MaxInstances: 16} +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 @@ -130,12 +130,21 @@ func TestExecuteChainRemovesRunStatusOnEarlyReturn(t *testing.T) { 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() + }, + }, { name: "failed to retrieve chain elements", ctx: context.Background, expect: func(mock pgxmock.PgxPoolIface) { mock.ExpectBegin() - mock.ExpectQuery("SELECT").WillReturnRows(pgxmock.NewRows([]string{"vxid"}).AddRow(int64(42))) + mock.ExpectQuery("FROM pg_locks").WillReturnRows(vxidRows()) mock.ExpectQuery("FROM timetable\\.task").WithArgs(c.ChainID).WillReturnError(errors.New("query_wait_timeout")) mock.ExpectRollback() }, @@ -146,13 +155,16 @@ func TestExecuteChainRemovesRunStatusOnEarlyReturn(t *testing.T) { t.Run(tc.name, func(t *testing.T) { mock, err := pgxmock.NewPool() assert.NoError(t, err) - pge := pgengine.NewDB(mock, "-c", clientName) + pge := pgengine.NewDB(mock, "-c", "scheduler_unit_test") sch := New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}), otel.NewNoop()) tc.expect(mock) - mock.ExpectExec("DELETE FROM timetable\\.active_chain"). - WithArgs(c.ChainID, clientName). - WillReturnResult(pgxmock.NewResult("DELETE", 1)) + // on_error handler must run even when the chain context is already cancelled + mock.ExpectBegin() + mock.ExpectQuery("FROM pg_locks").WillReturnRows(vxidRows()) + 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() sch.executeChain(tc.ctx(), c) assert.NoError(t, mock.ExpectationsWereMet()) @@ -160,6 +172,34 @@ func TestExecuteChainRemovesRunStatusOnEarlyReturn(t *testing.T) { } } +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 + go func() { + // let the worker pick up the chain, then shut the scheduler down + time.Sleep(100 * time.Millisecond) + cancel() + }() + sch.chainWorker(ctx, chains) + assert.NoError(t, mock.ExpectationsWereMet()) +} + func TestExecuteChainElement(t *testing.T) { mock, err := pgxmock.NewPool() assert.NoError(t, err) @@ -189,10 +229,12 @@ 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() + mock.ExpectBegin().WillReturnError(context.Canceled) 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..c3c51c17 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. From 1823969301a0a3acaf3debe0488ba80b9051fdfd Mon Sep 17 00:00:00 2001 From: olaberglund Date: Tue, 13 Oct 2026 09:42:10 +0200 Subject: [PATCH 3/6] [-] start the cleanup timeout when cleanup starts, not when the chain starts `executeChain` created its detached cleanup context before the chain's transaction began, so `DetachedTimeout` was counted from the start of the chain. Any chain running longer than 30s committed on an expired context: pgx never sent `COMMIT`, the chain's transaction was rolled back, and the chain was still logged as successful. Rollback, `on_error` and self-destruct after 30s hit the same expired context. Each exit path now creates its own detached context when its cleanup starts. `DetachedTimeout` is a variable so the new test can shorten it. Co-Authored-By: Claude Opus 5.5 --- internal/log/log.go | 2 +- internal/scheduler/chain.go | 24 +++++++++++-------- internal/scheduler/scheduler_test.go | 35 ++++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 11 deletions(-) diff --git a/internal/log/log.go b/internal/log/log.go index eb1b3c95..fa665896 100644 --- a/internal/log/log.go +++ b/internal/log/log.go @@ -121,7 +121,7 @@ func WithLogger(ctx context.Context, logger LoggerIface) context.Context { } // DetachedTimeout bounds cleanup work that must run even when the caller's context is cancelled. -const DetachedTimeout = 30 * time.Second +var DetachedTimeout = 30 * time.Second // 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. diff --git a/internal/scheduler/chain.go b/internal/scheduler/chain.go index 2a58611e..78488371 100644 --- a/internal/scheduler/chain.go +++ b/internal/scheduler/chain.go @@ -231,11 +231,16 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { } chainL := sch.l.WithField("chain", chain) - // cleanup (rollback, commit, on_error) must run even if ctx is cancelled, e.g. by notify_chain_stop() - bctx, bcancel := log.DetachedContext(log.WithLogger(ctx, chainL)) - defer bcancel() + cleanupContext := func(l log.LoggerIface) (context.Context, context.CancelFunc) { + return log.DetachedContext(log.WithLogger(ctx, l)) + } - chainFailed := func(msg string) { + chainFailed := func(l log.LoggerIface, tx pgx.Tx, msg string) { + bctx, cancel := cleanupContext(l) + defer cancel() + if tx != nil { + sch.pgengine.RollbackTransaction(bctx, tx) + } chainL.Error(msg) chainSpan.SetStatus(codes.Error, msg) sch.provider.RecordChainFailed(bctx, sch.Config().ClientName) @@ -245,17 +250,15 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { tx, vxid, err := sch.pgengine.StartTransaction(chainCtx) if err != nil { chainL.WithError(err).Error("Cannot start transaction") - chainFailed("Chain failed to start") + chainFailed(chainL, nil, "Chain failed to start") return } chainL = chainL.WithField("vxid", vxid) - bctx = log.WithLogger(bctx, chainL) err = sch.pgengine.GetChainElements(chainCtx, &ChainTasks, chain.ChainID) if err != nil { chainL.WithError(err).Error("Failed to retrieve chain elements") - sch.pgengine.RollbackTransaction(bctx, tx) - chainFailed("Chain failed to load tasks") + chainFailed(chainL, tx, "Chain failed to load tasks") return } @@ -275,13 +278,14 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { if err != nil { if !task.IgnoreError { - sch.pgengine.RollbackTransaction(log.WithLogger(bctx, l), tx) - chainFailed("Chain failed") + chainFailed(l, tx, "Chain failed") return } l.Info("Ignoring task failure") } } + bctx, bcancel := cleanupContext(chainL) + defer bcancel() sch.pgengine.CommitTransaction(bctx, tx) sch.provider.RecordChainDuration(ctx, time.Since(chainStart).Seconds(), sch.Config().ClientName) chainL.Info("Chain executed successfully") diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index c3c51c17..8cf4f508 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -53,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() From 8e2eaa08d2e5f5557d8dd99576206ac7b67df0bf Mon Sep 17 00:00:00 2001 From: olaberglund Date: Tue, 13 Oct 2026 09:48:03 +0200 Subject: [PATCH 4/6] [*] note why `DetachedTimeout` is a var --- internal/log/log.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/log/log.go b/internal/log/log.go index fa665896..a801347d 100644 --- a/internal/log/log.go +++ b/internal/log/log.go @@ -121,7 +121,7 @@ func WithLogger(ctx context.Context, logger LoggerIface) context.Context { } // DetachedTimeout bounds cleanup work that must run even when the caller's context is cancelled. -var DetachedTimeout = 30 * time.Second +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. From a7298147a402065c38bba13a197f7742dd6d0075 Mon Sep 17 00:00:00 2001 From: Pavlo Golub Date: Thu, 15 Oct 2026 17:30:40 +0200 Subject: [PATCH 5/6] [-] fix `RemoveChainRunStatus` dropping every instance's `active_chain` row --- internal/pgengine/access.go | 7 ++++++- internal/pgengine/pgengine_test.go | 18 +++++++++++++++--- 2 files changed, 21 insertions(+), 4 deletions(-) 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)) }) } From 259ae7a1987957d81f8a6b6c58e9a4bc8f77327e Mon Sep 17 00:00:00 2001 From: Pavlo Golub Date: Thu, 15 Oct 2026 17:30:40 +0200 Subject: [PATCH 6/6] [-] treat failed COMMIT as chain failure, skip `on_error` for stopped chains - commit has no deadline, `DetachedTimeout` bounds only rollback and `DeleteChain` - `on_error` runs on the chain context without a deadline - remove `CommitTransaction` wrapper, log early failures once --- internal/pgengine/transaction.go | 8 --- internal/pgengine/transaction_test.go | 8 +-- internal/scheduler/chain.go | 44 +++++++------- internal/scheduler/chain_test.go | 83 +++++++++++++++++++++------ 4 files changed, 88 insertions(+), 55 deletions(-) diff --git a/internal/pgengine/transaction.go b/internal/pgengine/transaction.go index 31ca4c20..6817dbb2 100644 --- a/internal/pgengine/transaction.go +++ b/internal/pgengine/transaction.go @@ -31,14 +31,6 @@ WHERE pid = pg_backend_pid() AND virtualxid IS NOT NULL`).Scan(&vxid) return tx, vxid, nil } -// 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") - } -} - // RollbackTransaction rollbacks transaction and log error in the case of error func (pge *PgEngine) RollbackTransaction(ctx context.Context, tx pgx.Tx) { err := tx.Rollback(ctx) diff --git a/internal/pgengine/transaction_test.go b/internal/pgengine/transaction_test.go index d8fa9eb5..75549b5d 100644 --- a/internal/pgengine/transaction_test.go +++ b/internal/pgengine/transaction_test.go @@ -65,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 78488371..7291731b 100644 --- a/internal/scheduler/chain.go +++ b/internal/scheduler/chain.go @@ -188,7 +188,7 @@ func (sch *Scheduler) removeChainRunStatus(ctx context.Context, chainID int) { } func (sch *Scheduler) executeOnErrorHandler(ctx context.Context, chain Chain) { - if chain.OnError == "" { + if ctx.Err() != nil || chain.OnError == "" { return } l := sch.l.WithField("chain", chain) @@ -200,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") @@ -231,34 +230,29 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { } chainL := sch.l.WithField("chain", chain) - cleanupContext := func(l log.LoggerIface) (context.Context, context.CancelFunc) { - return log.DetachedContext(log.WithLogger(ctx, l)) - } - - chainFailed := func(l log.LoggerIface, tx pgx.Tx, msg string) { - bctx, cancel := cleanupContext(l) - defer cancel() + 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() } - chainL.Error(msg) - chainSpan.SetStatus(codes.Error, msg) - sch.provider.RecordChainFailed(bctx, sch.Config().ClientName) - sch.executeOnErrorHandler(bctx, chain) + 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, "Chain failed to start") + 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") - chainFailed(chainL, tx, "Chain failed to load tasks") + chainFailed(chainL, tx, err, "Chain failed to load tasks") return } @@ -278,19 +272,23 @@ func (sch *Scheduler) executeChain(ctx context.Context, chain Chain) { if err != nil { if !task.IgnoreError { - chainFailed(l, tx, "Chain failed") + chainFailed(l, tx, err, "Chain failed") return } l.Info("Ignoring task failure") } } - bctx, bcancel := cleanupContext(chainL) - defer bcancel() - 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) 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 4caa5152..ce811d75 100644 --- a/internal/scheduler/chain_test.go +++ b/internal/scheduler/chain_test.go @@ -108,9 +108,10 @@ func TestExecuteChainEarlyReturn(t *testing.T) { 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) + name string + ctx func() context.Context + expect func(mock pgxmock.PgxPoolIface) + onError bool }{ { name: "cannot start transaction", @@ -118,9 +119,10 @@ func TestExecuteChainEarlyReturn(t *testing.T) { expect: func(mock pgxmock.PgxPoolIface) { mock.ExpectBegin().WillReturnError(errors.New("query_wait_timeout")) }, + onError: true, }, { - name: "cannot start transaction with cancelled context", + name: "chain stopped before start", ctx: func() context.Context { ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -138,6 +140,7 @@ func TestExecuteChainEarlyReturn(t *testing.T) { mock.ExpectQuery("FROM pg_locks").WillReturnError(errors.New("query_wait_timeout")) mock.ExpectRollback() }, + onError: true, }, { name: "failed to retrieve chain elements", @@ -148,6 +151,7 @@ func TestExecuteChainEarlyReturn(t *testing.T) { mock.ExpectQuery("FROM timetable\\.task").WithArgs(c.ChainID).WillReturnError(errors.New("query_wait_timeout")) mock.ExpectRollback() }, + onError: true, }, } @@ -159,19 +163,64 @@ func TestExecuteChainEarlyReturn(t *testing.T) { sch := New(pge, log.Init(config.LoggingOpts{LogLevel: "panic", LogDBLevel: "none"}), otel.NewNoop()) tc.expect(mock) - // on_error handler must run even when the chain context is already cancelled - mock.ExpectBegin() - mock.ExpectQuery("FROM pg_locks").WillReturnRows(vxidRows()) - 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() + if tc.onError { + expectOnError(mock) + } else { + mock.ExpectBegin() // must stay unconsumed: no on_error for a stopped chain + } sch.executeChain(tc.ctx(), c) - assert.NoError(t, mock.ExpectationsWereMet()) + 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} @@ -191,13 +240,14 @@ func TestChainWorkerRemovesRunStatusOnEarlyReturn(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) chains := make(chan Chain, 1) chains <- c + done := make(chan struct{}) go func() { - // let the worker pick up the chain, then shut the scheduler down - time.Sleep(100 * time.Millisecond) - cancel() + sch.chainWorker(ctx, chains) + close(done) }() - sch.chainWorker(ctx, chains) - assert.NoError(t, mock.ExpectationsWereMet()) + assert.Eventually(t, func() bool { return mock.ExpectationsWereMet() == nil }, 5*time.Second, 10*time.Millisecond) + cancel() + <-done } func TestExecuteChainElement(t *testing.T) { @@ -232,7 +282,6 @@ func TestExecuteOnErrorHandler(t *testing.T) { t.Run("check error handler if context cancelled", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - mock.ExpectBegin().WillReturnError(context.Canceled) sch.executeOnErrorHandler(ctx, c) assert.NoError(t, mock.ExpectationsWereMet()) })