From 65164ba8a0de8ca11912be8011076c07087cbdfa Mon Sep 17 00:00:00 2001 From: Pavlo Golub Date: Wed, 15 Jul 2026 17:21:34 +0200 Subject: [PATCH] [+] add ability to enable/disable tasks, closes #792 --- .github/agents/chainer.agent.md | 1 + docs/components.md | 1 + docs/yaml-format.md | 2 + internal/pgengine/access.go | 2 +- internal/pgengine/bootstrap_test.go | 4 +- internal/pgengine/migration.go | 6 ++ internal/pgengine/pgengine_test.go | 16 ++++ internal/pgengine/sql/ddl.sql | 5 +- internal/pgengine/sql/init.sql | 3 +- internal/pgengine/sql/migrations/00792.sql | 4 + internal/pgengine/yaml.go | 13 +++- internal/pgengine/yaml_test.go | 86 ++++++++++++++++++---- main.go | 2 +- 13 files changed, 120 insertions(+), 25 deletions(-) create mode 100644 internal/pgengine/sql/migrations/00792.sql diff --git a/.github/agents/chainer.agent.md b/.github/agents/chainer.agent.md index 8c4fa933..dc044a52 100644 --- a/.github/agents/chainer.agent.md +++ b/.github/agents/chainer.agent.md @@ -74,6 +74,7 @@ Stores individual tasks within chains. | `ignore_error` | BOOLEAN | Continue chain even if task fails | | `autonomous` | BOOLEAN | Execute outside chain transaction | | `timeout` | INTEGER | Task timeout in milliseconds (0 = no timeout) | +| `live` | BOOLEAN | Task is executed only when TRUE (default), set FALSE to skip | #### Table: timetable.parameter diff --git a/docs/components.md b/docs/components.md index 485cad00..9067ab6c 100644 --- a/docs/components.md +++ b/docs/components.md @@ -62,6 +62,7 @@ The next building block is a **task**, which simply represents a step in a list | `ignore_error` | `boolean` | Specify if the next task should proceed after encountering an error (default: `false`) | | `autonomous` | `boolean` | Specify if the task should be executed out of the chain transaction. Useful for `VACUUM`, `CREATE DATABASE`, `CALL` etc. | | `timeout` | `integer` | Abort any task within a chain that takes more than the specified number of milliseconds | +| `live` | `boolean` | Indication that the task is ready to run, set to `false` to skip execution (default: `true`) | !!! warning diff --git a/docs/yaml-format.md b/docs/yaml-format.md index d6c90354..aa66530c 100644 --- a/docs/yaml-format.md +++ b/docs/yaml-format.md @@ -29,6 +29,7 @@ chains: ignore_error: false # Optional: ignore_error (BOOLEAN), default: false autonomous: false # Optional: autonomous (BOOLEAN), default: false timeout: 5000 # Optional: timeout in milliseconds (INTEGER) + live: true # Optional: live (BOOLEAN), default: true; set false to skip the task - name: "task-2" kind: "PROGRAM" @@ -66,6 +67,7 @@ chains: | `ignore_error` | `ignore_error` | BOOLEAN | `false` | Continue on error | | `autonomous` | `autonomous` | BOOLEAN | `false` | Execute outside transaction | | `timeout` | `timeout` | INTEGER | `0` | Task timeout (ms) | +| `live` | `live` | BOOLEAN | `true` | Whether task is executed; disabled tasks are skipped | ## Task Ordering diff --git a/internal/pgengine/access.go b/internal/pgengine/access.go index 37bc29ac..b9e93124 100644 --- a/internal/pgengine/access.go +++ b/internal/pgengine/access.go @@ -142,7 +142,7 @@ func (pge *PgEngine) GetChainElements(ctx context.Context, chainTasks *[]ChainTa autonomous, COALESCE(database_connection, '') as database_connection, timeout -FROM timetable.task WHERE chain_id = $1 ORDER BY task_order ASC` +FROM timetable.task WHERE chain_id = $1 AND live ORDER BY task_order ASC` rows, err := pge.ConfigDb.Query(ctx, sqlSelectChainTasks, chainID) if err != nil { return err diff --git a/internal/pgengine/bootstrap_test.go b/internal/pgengine/bootstrap_test.go index 461a7231..d2f2e5e8 100644 --- a/internal/pgengine/bootstrap_test.go +++ b/internal/pgengine/bootstrap_test.go @@ -255,7 +255,7 @@ func TestExecuteFileScript(t *testing.T) { WithArgs(anyArgs(9)...). WillReturnRows(pgxmock.NewRows([]string{"chain_id"}).AddRow(1)) mockPool.ExpectQuery("INSERT INTO timetable\\.task"). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) err = mockpge.ExecuteFileScript(context.Background(), cmdOpts, yamlFile) @@ -333,7 +333,7 @@ func TestExecuteFileScript(t *testing.T) { WithArgs(anyArgs(9)...). WillReturnRows(pgxmock.NewRows([]string{"chain_id"}).AddRow(1)) mockPool.ExpectQuery("INSERT INTO timetable\\.task"). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) err = mockpge.ExecuteFileScript(context.Background(), cmdOpts, yamlFile) diff --git a/internal/pgengine/migration.go b/internal/pgengine/migration.go index 6486480c..e3af90a5 100644 --- a/internal/pgengine/migration.go +++ b/internal/pgengine/migration.go @@ -153,6 +153,12 @@ var Migrations func() migrator.Option = func() migrator.Option { return ExecuteMigrationScript(ctx, tx, "00733.sql") }, }, + &migrator.Migration{ + Name: "00792 Add ability to enable and disable tasks", + Func: func(ctx context.Context, tx pgx.Tx) error { + return ExecuteMigrationScript(ctx, tx, "00792.sql") + }, + }, // adding new migration here, update "timetable"."migration" in "sql/init.sql" // and "dbapi" variable in main.go! diff --git a/internal/pgengine/pgengine_test.go b/internal/pgengine/pgengine_test.go index 84773e3d..e9a68d06 100644 --- a/internal/pgengine/pgengine_test.go +++ b/internal/pgengine/pgengine_test.go @@ -112,6 +112,22 @@ func TestSchedulerFunctions(t *testing.T) { pge.CommitTransaction(ctx, tx) }) + t.Run("Check GetChainElements returns only live tasks", func(t *testing.T) { + var chainID int + err := pge.ConfigDb.QueryRow(ctx, + `INSERT INTO timetable.chain (chain_name, run_at) VALUES ('live_flag_chain', '@reboot') RETURNING chain_id`).Scan(&chainID) + assert.NoError(t, err, "Should insert chain") + _, err = pge.ConfigDb.Exec(ctx, + `INSERT INTO timetable.task (chain_id, task_order, command, live) VALUES ($1, 10, 'SELECT 1', TRUE), ($1, 20, 'SELECT 2', FALSE)`, chainID) + assert.NoError(t, err, "Should insert tasks") + var tasks []pgengine.ChainTask + assert.NoError(t, pge.GetChainElements(ctx, &tasks, chainID), "Should return tasks") + assert.Len(t, tasks, 1, "Disabled task should be filtered out") + assert.Equal(t, "SELECT 1", tasks[0].Command, "Only the live task should be returned") + _, err = pge.ConfigDb.Exec(ctx, `DELETE FROM timetable.chain WHERE chain_id = $1`, chainID) + assert.NoError(t, err) + }) + t.Run("Check GetChainParamValues function", func(t *testing.T) { var paramVals []string tx, txid, err := pge.StartTransaction(ctx) diff --git a/internal/pgengine/sql/ddl.sql b/internal/pgengine/sql/ddl.sql index 00ef3d49..10322272 100644 --- a/internal/pgengine/sql/ddl.sql +++ b/internal/pgengine/sql/ddl.sql @@ -41,7 +41,8 @@ CREATE TABLE timetable.task ( database_connection TEXT, ignore_error BOOLEAN NOT NULL DEFAULT FALSE, autonomous BOOLEAN NOT NULL DEFAULT FALSE, - timeout INTEGER DEFAULT 0 + timeout INTEGER DEFAULT 0, + live BOOLEAN NOT NULL DEFAULT TRUE ); COMMENT ON TABLE timetable.task IS @@ -62,6 +63,8 @@ COMMENT ON COLUMN timetable.task.timeout IS 'Abort any task within a chain that takes more than the specified number of milliseconds'; COMMENT ON COLUMN timetable.task.autonomous IS 'Specify if the task should be executed out of the chain transaction. Useful for VACUUM, CREATE DATABASE, CALL etc.'; +COMMENT ON COLUMN timetable.task.live IS + 'Indication that the task is ready to run, set to FALSE to skip execution'; -- parameter passing for a chain task CREATE TABLE timetable.parameter( diff --git a/internal/pgengine/sql/init.sql b/internal/pgengine/sql/init.sql index 97572c76..19412df3 100644 --- a/internal/pgengine/sql/init.sql +++ b/internal/pgengine/sql/init.sql @@ -28,4 +28,5 @@ VALUES (12, '00575 Add on_error handling'), (13, '00629 Add ignore_error column to timetable.execution_log'), (14, '00721 Add more job control functions'), - (15, '00733 Add params column to timetable.execution_log table'); \ No newline at end of file + (15, '00733 Add params column to timetable.execution_log table'), + (16, '00792 Add ability to enable and disable tasks'); \ No newline at end of file diff --git a/internal/pgengine/sql/migrations/00792.sql b/internal/pgengine/sql/migrations/00792.sql new file mode 100644 index 00000000..d2e21061 --- /dev/null +++ b/internal/pgengine/sql/migrations/00792.sql @@ -0,0 +1,4 @@ +ALTER TABLE timetable.task ADD COLUMN live BOOLEAN NOT NULL DEFAULT TRUE; + +COMMENT ON COLUMN timetable.task.live IS + 'Indication that the task is ready to run, set to FALSE to skip execution'; diff --git a/internal/pgengine/yaml.go b/internal/pgengine/yaml.go index 92a63c73..9ef4b996 100644 --- a/internal/pgengine/yaml.go +++ b/internal/pgengine/yaml.go @@ -23,6 +23,7 @@ type YamlChain struct { type YamlTask struct { ChainTask `yaml:",inline"` TaskName string `db:"task_name" yaml:"name,omitempty"` + Live *bool `yaml:"live,omitempty"` Parameters []any `yaml:"parameters,omitempty"` } @@ -100,8 +101,8 @@ func (pge *PgEngine) CreateChainFromYaml(ctx context.Context, yamlChain *YamlCha err := pge.ConfigDb.QueryRow(ctx, ` INSERT INTO timetable.task ( chain_id, task_order, task_name, kind, command, - run_as, database_connection, ignore_error, autonomous, timeout - ) VALUES ($1, $2, $3, $4::timetable.command_kind, $5, $6, $7, $8, $9, $10) + run_as, database_connection, ignore_error, autonomous, timeout, live + ) VALUES ($1, $2, $3, $4::timetable.command_kind, $5, $6, $7, $8, $9, $10, $11) RETURNING task_id`, chainID, taskOrder, @@ -112,7 +113,8 @@ func (pge *PgEngine) CreateChainFromYaml(ctx context.Context, yamlChain *YamlCha nullString(task.ConnectString), task.IgnoreError, task.Autonomous, - task.Timeout).Scan(&taskID) + task.Timeout, + task.Live == nil || *task.Live).Scan(&taskID) if err != nil { return 0, fmt.Errorf("failed to insert task %d: %w", i+1, err) } @@ -225,6 +227,11 @@ func (c *YamlChain) SetDefaults() { if task.Kind == "" { task.Kind = "SQL" } + // Tasks are live by default, unlike chains + if task.Live == nil { + live := true + task.Live = &live + } } } diff --git a/internal/pgengine/yaml_test.go b/internal/pgengine/yaml_test.go index 9fc1d9f6..1c1abf4a 100644 --- a/internal/pgengine/yaml_test.go +++ b/internal/pgengine/yaml_test.go @@ -108,6 +108,35 @@ func TestLoadYamlChainsIntegration(t *testing.T) { require.NoError(t, err) assert.Equal(t, "0 1 * * *", schedule) }) + + t.Run("Task live flag", func(t *testing.T) { + yamlContent := `chains: + - name: test-task-live + schedule: "0 0 * * *" + tasks: + - name: enabled-by-default + command: SELECT 1 + - name: explicitly-enabled + command: SELECT 2 + live: true + - name: disabled + command: SELECT 3 + live: false` + + tempFile := createTempYamlFile(t, yamlContent) + defer removeTempFile(t, tempFile) + + err := pge.LoadYamlChains(ctx, tempFile, false) + require.NoError(t, err) + + rows, err := pge.ConfigDb.Query(ctx, + `SELECT t.live FROM timetable.task t JOIN timetable.chain c ON c.chain_id = t.chain_id + WHERE c.chain_name = $1 ORDER BY t.task_order`, "test-task-live") + require.NoError(t, err) + lives, err := pgx.CollectRows(rows, pgx.RowTo[bool]) + require.NoError(t, err) + assert.Equal(t, []bool{true, true, false}, lives) + }) } func TestYamlParameterHandling(t *testing.T) { @@ -521,6 +550,9 @@ func TestYamlChainSetDefaults(t *testing.T) { chain.SetDefaults() assert.Equal(t, "* * * * *", chain.Schedule) assert.Equal(t, "SQL", chain.Tasks[0].Kind) + if assert.NotNil(t, chain.Tasks[0].Live) { + assert.True(t, *chain.Tasks[0].Live, "Task should be live by default") + } }) t.Run("Keep existing values", func(t *testing.T) { @@ -543,6 +575,28 @@ func TestYamlChainSetDefaults(t *testing.T) { assert.Equal(t, "0 0 * * *", chain.Schedule) assert.Equal(t, "PROGRAM", chain.Tasks[0].Kind) }) + + t.Run("Keep disabled task", func(t *testing.T) { + disabled := false + chain := &pgengine.YamlChain{ + Chain: pgengine.Chain{ + ChainName: "test-chain", + }, + Tasks: []pgengine.YamlTask{ + { + ChainTask: pgengine.ChainTask{ + Command: "SELECT 1", + }, + Live: &disabled, + }, + }, + } + + chain.SetDefaults() + if assert.NotNil(t, chain.Tasks[0].Live) { + assert.False(t, *chain.Tasks[0].Live, "Task should stay disabled") + } + }) } func TestParameterStorageIntegration(t *testing.T) { @@ -713,7 +767,7 @@ func TestNullString(t *testing.T) { WithArgs(anyArgs(9)...). WillReturnRows(pgxmock.NewRows([]string{"chain_id"}).AddRow(1)) mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) err := mockpge.LoadYamlChains(context.Background(), tmpfile, false) @@ -769,7 +823,7 @@ func TestLoadYamlChainsMultiTask(t *testing.T) { // Mock first task creation mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) // Mock first task parameters (2 parameters) mockPool.ExpectExec(`INSERT INTO timetable\.parameter`). @@ -779,7 +833,7 @@ func TestLoadYamlChainsMultiTask(t *testing.T) { // Mock second task creation mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(2)) // Mock second task parameters (2 parameters) mockPool.ExpectExec(`INSERT INTO timetable\.parameter`). @@ -891,11 +945,11 @@ func TestCreateChainFromYamlEdgeCases(t *testing.T) { // Mock first task creation (no parameters) mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) // Mock second task creation (empty parameters) mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(2)) err := mockpge.LoadYamlChains(context.Background(), tmpfile, false) @@ -925,7 +979,7 @@ func TestCreateChainFromYamlEdgeCases(t *testing.T) { WithArgs(anyArgs(9)...). WillReturnRows(pgxmock.NewRows([]string{"chain_id"}).AddRow(1)) mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) // Mock parameter insertion mockPool.ExpectExec(`INSERT INTO timetable\.parameter`). @@ -969,7 +1023,7 @@ func TestCreateChainFromYamlErrorHandling(t *testing.T) { // Mock first task with complex parameter mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) mockPool.ExpectExec(`INSERT INTO timetable\.parameter`). WithArgs(anyArgs(3)...). @@ -977,7 +1031,7 @@ func TestCreateChainFromYamlErrorHandling(t *testing.T) { // Mock second task (no parameters) mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(2)) err := mockpge.LoadYamlChains(context.Background(), tmpfile, false) @@ -1035,7 +1089,7 @@ func TestCreateChainFromYamlErrorHandling(t *testing.T) { // Mock sql-task creation with 2 parameters mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) mockPool.ExpectExec(`INSERT INTO timetable\.parameter`). WithArgs(anyArgs(3)...). @@ -1044,7 +1098,7 @@ func TestCreateChainFromYamlErrorHandling(t *testing.T) { // Mock program-task creation with 2 parameters mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(2)) mockPool.ExpectExec(`INSERT INTO timetable\.parameter`). WithArgs(anyArgs(3)...). @@ -1053,7 +1107,7 @@ func TestCreateChainFromYamlErrorHandling(t *testing.T) { // Mock builtin-task creation with 1 parameter mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(3)) mockPool.ExpectExec(`INSERT INTO timetable\.parameter`). WithArgs(anyArgs(3)...). @@ -1096,7 +1150,7 @@ func TestNullStringFunction(t *testing.T) { WillReturnRows(pgxmock.NewRows([]string{"chain_id"}).AddRow(1)) // Mock task creation with NULL fields mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) err := mockpge.LoadYamlChains(context.Background(), tmpfile, false) @@ -1129,7 +1183,7 @@ func TestNullStringFunction(t *testing.T) { WillReturnRows(pgxmock.NewRows([]string{"chain_id"}).AddRow(1)) // Mock task creation with mixed NULL/non-NULL fields mockPool.ExpectQuery(`INSERT INTO timetable\.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) err := mockpge.LoadYamlChains(context.Background(), tmpfile, false) @@ -1157,7 +1211,7 @@ func TestCreateChainFromYamlErrors(t *testing.T) { WithArgs(anyArgs(9)...). WillReturnRows(pgxmock.NewRows([]string{"chain_id"}).AddRow(1)) mockPool.ExpectQuery(`INSERT INTO timetable.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnError(fmt.Errorf("simulated DB error on task")) _, err := mockpge.CreateChainFromYaml(ctx, &pgengine.YamlChain{ @@ -1176,7 +1230,7 @@ func TestCreateChainFromYamlErrors(t *testing.T) { WithArgs(anyArgs(9)...). WillReturnRows(pgxmock.NewRows([]string{"chain_id"}).AddRow(1)) mockPool.ExpectQuery(`INSERT INTO timetable.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) _, err := mockpge.CreateChainFromYaml(ctx, &pgengine.YamlChain{ @@ -1198,7 +1252,7 @@ func TestCreateChainFromYamlErrors(t *testing.T) { WithArgs(anyArgs(9)...). WillReturnRows(pgxmock.NewRows([]string{"chain_id"}).AddRow(1)) mockPool.ExpectQuery(`INSERT INTO timetable.task`). - WithArgs(anyArgs(10)...). + WithArgs(anyArgs(11)...). WillReturnRows(pgxmock.NewRows([]string{"task_id"}).AddRow(1)) mockPool.ExpectExec(`INSERT INTO timetable.parameter`). WithArgs(anyArgs(3)...). diff --git a/main.go b/main.go index d304706f..481d2726 100644 --- a/main.go +++ b/main.go @@ -55,7 +55,7 @@ var ( commit = "000000" version = "master" date = "unknown" - dbapi = "00733" + dbapi = "00792" ) func printVersion() {