[+] add ability to enable/disable tasks, closes #792

.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
 

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
 

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
 

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

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)

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!
 

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)

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(

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');
+    (15, '00733 Add params column to timetable.execution_log table'),
+    (16, '00792 Add ability to enable and disable tasks');

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';

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
+		}
 	}
 }
 

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,15 +1023,15 @@ 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)...).
 			WillReturnResult(pgxmock.NewResult("INSERT", 1))
 
 		// 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)...).

main.go

@@ -55,7 +55,7 @@ var (
 	commit  = "000000"
 	version = "master"
 	date    = "unknown"
-	dbapi   = "00733"
+	dbapi   = "00792"
 )
 
 func printVersion() {