feat: scoped tasks #248

jason-lynch · 2026-01-14T22:43:15Z

Summary

Adds a "scope" concept to tasks to support tasks that apply to non-database entities. With this change, the task's parent is identified by the scope and entity_id fields.

This changes how tasks and task logs are stored, so this PR includes a non-destructive, idempotent migration to move tasks and task logs to a new keyspace.

Changes

Adds scope and entity_id fields to tasks and task logs
Migrates existing tasks and task logs to a new keyspace so that they're still available via the API

Testing

Start a server on the main branch
Create a database
Switch to this branch
Restart the server
Verify that existing tasks and task logs are still available via the API
Create another database
Verify that new tasks and tasks logs are available via the API
Verify that the scope and entity_id fields exist and are set correctly on all tasks

PLAT-347

Summary by CodeRabbit

New Features
- Added support for non-database tasks with scope-based identification. Tasks can now be associated with different entity types (databases and hosts) using scope and entity_id fields.
- Extended task tracking to support per-entity scope identifiers, enabling better organization of database and host-related tasks.
Refactor
- Migrated task identification system from database_id-only to scope + entity_id approach for improved multi-entity support.
Tests
- Added comprehensive test coverage for scoped task operations across databases and hosts.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

coderabbitai · 2026-01-14T22:43:20Z

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

📝 Walkthrough

Walkthrough

This pull request extends task-related API types and infrastructure to support multi-tenant scope and entity identification. Tasks now carry scope and entity_id fields alongside database_id, enabling support for non-database tasks (e.g., host-scoped tasks). Required fields in API payloads shift from database_id to scope and entity_id.

Changes

Cohort / File(s)	Summary
API Design Layer `api/apiv1/design/task.go`	Added public attributes `scope` and `entity_id` to Task, TaskLog, and ListDatabaseTasksResponse types. Updated required fields from `database_id` to `scope` and `entity_id`. Updated example payloads to reflect new fields. Marked `database_id` as deprecated in TaskLog.
Domain Model `server/internal/task/task.go`, `task_scope_test.go`	Introduced `Scope` type with constants `ScopeDatabase` and `ScopeHost`. Extended `Task` and `TaskLog` structs with `Scope` and `EntityID` fields. Added `Options.EntityID()` method. Enhanced validation to enforce scope-specific required IDs. Added test coverage for scope validation and field behavior.
Changelog `changes/unreleased/Added-20260114-173755.yaml`	Added release notes documenting support for non-database tasks and new scope/entity_id fields.
Storage Layer `server/internal/task/task_store.go`, `task_store_test.go`	Refactored key generation from `(databaseID, taskID)` to `(scope, entityID, taskID)`. Changed prefix from `"tasks"` to `"tasks_v2"`. Added `EntityPrefix()` and updated `GetAllByDatabaseID` → `GetAllByEntity()`, `Delete()`, and `DeleteByDatabaseID()` → `DeleteByEntity()`. Added comprehensive test suite covering scope-aware CRUD operations.
Task Log Storage `server/internal/task/task_log_store.go`, `task_log_store_test.go`	Added `Scope` and `EntityID` fields to `StoredTaskLogEntry`. Renamed storage prefix to `"task_log_entries"`. Refactored methods to use scope-based keys. Added `TaskPrefix()`, `Key()`, `GetAllByTask()`, `DeleteByTask()`, and `DeleteByEntity()`. Deprecated legacy database-ID methods. Added extensive test coverage for scope-aware storage operations.
Service Layer `server/internal/task/service.go`, `service_scoped_test.go`	Updated all public methods (`GetTask`, `GetTasks`, `DeleteTask`, `AddLogEntry`, `GetTaskLog`, etc.) to accept `scope` and `entityID` parameters instead of `databaseID`. Reworked storage key resolution and paging logic. Updated task log persistence to include scope/entity fields. Added comprehensive integration test suite validating scope-aware task lifecycle.
Conversion & Serialization `server/internal/api/apiv1/convert.go`	Updated `taskToAPI()` and `taskLogToAPI()` to populate `Scope` and `EntityID` fields in API responses. Made `DatabaseID` nullable.
Task Log Writer `server/internal/task/task_log_writer.go`	Updated `NewTaskLogWriter()` constructor signature to accept `scope` and `entityID` parameters instead of `databaseID`.
Migration Infrastructure `server/internal/migrate/migrations/add_task_scope.go`, `all_migrations.go`	Added new `AddTaskScope` migration to convert legacy etcd-based tasks to new scope-aware storage model. Reads old task/task_log entries, converts to new format with scope/entity fields, and writes to new stores.
API Handlers `server/internal/api/apiv1/post_init_handlers.go`	Updated handlers (`ListDatabaseTasks`, `GetDatabaseTask`, `GetTaskLog`, etc.) to pass `task.ScopeDatabase` scope parameter to task service methods.
Orchestration `server/internal/orchestrator/swarm/pgbackrest_restore.go`	Updated `GetTask()` and `NewTaskLogWriter()` calls to include `task.ScopeDatabase` scope parameter.
Workflow Activities `server/internal/workflows/activities/apply_event.go`, `create_pgbackrest_backup.go`, `log_task_event.go`, `update_task.go`	Updated activity input structures and task service calls to use scope/entityID instead of databaseID. Renamed internal logging function. Updated all task log writer and service invocations with scope parameter.
Workflow Orchestration `server/internal/workflows/common.go`, `service.go`	Updated `logTaskEvent()` and `cancelTask()` method signatures to accept scope and entityID. Updated all workflow files (`create_pgbackrest_backup.go`, `delete_database.go`, `failover.go`, `pgbackrest_restore.go`, `refresh_current_state.go`, `restart_instance.go`, `start_instance.go`, `stop_instance.go`, `switchover.go`, `update_database.go`) to construct `UpdateTaskInput` with `Scope` and `EntityID` fields instead of `DatabaseID`. Consistently pass `task.ScopeDatabase` in scope-aware task operations.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Poem

🐰 Hops through scopes and entities with glee,
Tasks now dance in tenants wild and free,
DatabaseID steps aside with grace,
Scope and EntityID take their place! ✨

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 4.76% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title 'feat: scoped tasks' clearly and concisely summarizes the main change: introducing a scope concept for tasks to support non-database entities.
Description check	✅ Passed	The PR description includes all required sections: a clear summary explaining the scope concept, a detailed changes list, comprehensive testing instructions, and a checklist confirming added tests and changelog entry.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

jason-lynch · 2026-01-15T17:37:14Z

@coderabbitai review

coderabbitai · 2026-01-15T17:37:20Z

✅ Actions performed

Review triggered.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

coderabbitai

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (4)

server/internal/orchestrator/swarm/pgbackrest_restore.go (1)
94-97: Bug: handleError returns wrong error variable.

The closure returns err (the error from startTask on line 93) instead of cause (the parameter passed to handleError). This means all error returns from handleError calls will return the original startTask error rather than the actual failure cause.
🐛 Proposed fix
 	handleError := func(cause error) error {
 		p.failTask(logger, taskSvc, t, cause)
-		return err
+		return cause
 	}
server/internal/task/service.go (2)
274-286: Bug: Empty Statuses slice filters out all tasks.

When opts.Statuses is empty, slices.Contains(opts.Statuses, task.Status) always returns false, causing all tasks to be filtered out. The check should only apply when statuses are specified.
Proposed fix
 func matchesFilters(task *Task, opts TaskListOptions) bool {
 	if opts.Type != "" && task.Type != opts.Type {
 		return false
 	}
-	if !slices.Contains(opts.Statuses, task.Status) {
+	if len(opts.Statuses) > 0 && !slices.Contains(opts.Statuses, task.Status) {
 		return false
 	}
 	if opts.NodeName != "" && (task == nil || task.NodeName != opts.NodeName) {
 		return false
 	}

 	return true
 }
118-127: Conditional assignment for DatabaseID based on scope.

Setting DatabaseID unconditionally to entityID will store incorrect data when the scope is not ScopeDatabase (e.g., for host-scoped tasks, DatabaseID would contain a host ID).
Proposed fix
 	stored := &StoredTaskLogEntry{
 		Scope:      scope,
 		EntityID:   entityID,
-		DatabaseID: entityID, // For backward compatibility with database-scoped tasks
+		DatabaseID: "", // Set below for backward compatibility
 		TaskID:     taskID,
 		EntryID:    entryID,
 		Timestamp:  timestamp,
 		Message:    entry.Message,
 		Fields:     entry.Fields,
 	}
+	if scope == ScopeDatabase {
+		stored.DatabaseID = entityID
+	}
server/internal/workflows/common.go (1)
176-196: Bug: Error logged unconditionally regardless of logTaskEvent outcome.

Line 194 logs an error message even when err is nil, resulting in misleading log entries on successful task event logging.
🐛 Proposed fix
 	err := w.logTaskEvent(cleanupCtx, scope, entityID, taskID, task.LogEntry{
 		Message: "task successfully canceled",
 		Fields:  map[string]any{"status": "canceled"},
 	})
-	logger.With("error", err).Error("failed to log task event")
+	if err != nil {
+		logger.With("error", err).Error("failed to log task event")
+	}
 
 }

🤖 Fix all issues with AI agents

In `@server/internal/migrate/migrations/add_task_scope.go`:
- Around line 66-71: The error message for the Etcd query uses the wrong
wording: change the fmt.Errorf call that currently returns "failed to query for
old tasks: %w" (in the block using oldTaskLogsPrefix / oldTaskLogsRangeOp /
oldTaskLogs) to a correct message like "failed to query for old task logs: %w"
so the log accurately reflects the operation querying oldStoredTaskLogEntry
entries.

🧹 Nitpick comments (7)

server/internal/task/task_log_writer.go (1)
12-16: Consider renaming DatabaseID to EntityID for clarity.

The TaskLogWriter struct field DatabaseID now holds the generic entityID value. While this maintains backward compatibility, it could cause confusion since the field name no longer accurately describes its contents for non-database scopes (e.g., host-scoped tasks).

If backward compatibility is essential, consider adding a comment explaining this or introducing an EntityID field alongside for future use.
♻️ Optional: Add clarifying comment or alias
 type TaskLogWriter struct {
-	DatabaseID string
+	DatabaseID string // Legacy name; holds entityID for all scopes
 	TaskID     uuid.UUID
 	writer     *utils.LineWriter
 }
Or for a cleaner approach in a future refactor:
 type TaskLogWriter struct {
-	DatabaseID string
+	EntityID   string
 	TaskID     uuid.UUID
 	writer     *utils.LineWriter
 }
Also applies to: 18-31
server/internal/task/task_scope_test.go (1)
26-35: Consider adding host scope test case.

The test validates EntityID() for database scope. Consider adding a similar test for host scope to ensure complete coverage of the EntityID() method's switch statement.
♻️ Suggested addition
func TestOptionsWithScope(t *testing.T) {
	t.Run("database scope", func(t *testing.T) {
		opts := task.Options{
			Scope:      task.ScopeDatabase,
			DatabaseID: "my-database",
			Type:       task.TypeCreate,
		}
		assert.Equal(t, task.ScopeDatabase, opts.Scope)
		assert.Equal(t, "my-database", opts.EntityID())
	})

	t.Run("host scope", func(t *testing.T) {
		opts := task.Options{
			Scope:  task.ScopeHost,
			HostID: "host-1",
			Type:   task.TypeRemoveHost,
		}
		assert.Equal(t, task.ScopeHost, opts.Scope)
		assert.Equal(t, "host-1", opts.EntityID())
	})
}
server/internal/migrate/migrations/add_task_scope.go (1)
73-78: Consider adding idempotency logging for task logs.

For consistency with the task migration pattern (lines 54-64), consider checking for ErrAlreadyExists on task log migration and logging a skip message. This would provide better observability during repeated migration runs.
Proposed change
 	for _, oldTaskLog := range oldTaskLogs {
-		err := taskStore.TaskLogMessage.Put(oldTaskLog.convert()).Exec(ctx)
-		if err != nil {
+		err := taskStore.TaskLogMessage.Create(oldTaskLog.convert()).Exec(ctx)
+		switch {
+		case errors.Is(err, storage.ErrAlreadyExists):
+			logger.Info().
+				Stringer("entry_id", oldTaskLog.EntryID).
+				Stringer("task_id", oldTaskLog.TaskID).
+				Msg("task log entry has already been migrated, skipping")
+		case err != nil:
 			return fmt.Errorf("failed to migrate task log entry %s for task %s: %w", oldTaskLog.EntryID, oldTaskLog.TaskID, err)
 		}
 	}
Note: This assumes TaskLogMessage has a Create method similar to tasks. If only Put is available, the current approach is acceptable since the data is identical on re-runs.
server/internal/workflows/activities/log_task_event.go (1)
37-46: Update logger key to reflect scope-agnostic entity.

The logger key "database_id" is misleading for non-database scopes (e.g., host-scoped tasks). Consider updating to "entity_id" for consistency with the new scoped model.
Proposed fix
 func (a *Activities) LogTaskEvent(ctx context.Context, input *LogTaskEventInput) (*LogTaskEventOutput, error) {
-	logger := activity.Logger(ctx).With("database_id", input.EntityID)
+	logger := activity.Logger(ctx).With("scope", input.Scope, "entity_id", input.EntityID)
 	logger.Debug("logging task event")
server/internal/api/apiv1/convert.go (1)
568-572: Consider using consistent Scope conversion.

In taskToAPI (line 539), Scope is converted using string(t.Scope), while here it uses t.Scope.String(). Both work since Scope is a string type alias, but using the same pattern improves consistency.
♻️ Optional: Use consistent conversion pattern
 return &api.TaskLog{
-		Scope:       t.Scope.String(),
+		Scope:       string(t.Scope),
 		EntityID:    t.EntityID,
api/apiv1/design/task.go (2)
13-16: Consider adding Enum constraint for scope.

The description states scope can be "database or host", but there's no validation constraint. Adding an enum would prevent invalid values and provide better API documentation.
♻️ Suggested improvement
 g.Attribute("scope", g.String, func() {
+    g.Enum("database", "host")
     g.Description("The scope of the task (database or host).")
     g.Example("database")
 })
101-104: Add Enum constraint for consistency with Task type.

Same recommendation as for the Task type—adding g.Enum("database", "host") would ensure validation consistency across both types.
♻️ Suggested improvement
 g.Attribute("scope", g.String, func() {
+    g.Enum("database", "host")
     g.Description("The scope of the task (database or host).")
     g.Example("database")
 })

📜 Review details

Configuration used: defaults

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between d35513b and 53678c0.

⛔ Files ignored due to path filters (10)

api/apiv1/gen/control_plane/service.go is excluded by !**/gen/**
api/apiv1/gen/control_plane/views/view.go is excluded by !**/gen/**
api/apiv1/gen/http/control_plane/client/encode_decode.go is excluded by !**/gen/**
api/apiv1/gen/http/control_plane/client/types.go is excluded by !**/gen/**
api/apiv1/gen/http/control_plane/server/encode_decode.go is excluded by !**/gen/**
api/apiv1/gen/http/control_plane/server/types.go is excluded by !**/gen/**
api/apiv1/gen/http/openapi.json is excluded by !**/gen/**
api/apiv1/gen/http/openapi.yaml is excluded by !**/gen/**
api/apiv1/gen/http/openapi3.json is excluded by !**/gen/**
api/apiv1/gen/http/openapi3.yaml is excluded by !**/gen/**

📒 Files selected for processing (32)

api/apiv1/design/task.go
changes/unreleased/Added-20260114-173755.yaml
server/internal/api/apiv1/convert.go
server/internal/api/apiv1/post_init_handlers.go
server/internal/migrate/all_migrations.go
server/internal/migrate/migrations/add_task_scope.go
server/internal/orchestrator/swarm/pgbackrest_restore.go
server/internal/task/service.go
server/internal/task/service_scoped_test.go
server/internal/task/task.go
server/internal/task/task_log_store.go
server/internal/task/task_log_store_test.go
server/internal/task/task_log_writer.go
server/internal/task/task_scope_test.go
server/internal/task/task_store.go
server/internal/task/task_store_test.go
server/internal/workflows/activities/apply_event.go
server/internal/workflows/activities/create_pgbackrest_backup.go
server/internal/workflows/activities/log_task_event.go
server/internal/workflows/activities/update_task.go
server/internal/workflows/common.go
server/internal/workflows/create_pgbackrest_backup.go
server/internal/workflows/delete_database.go
server/internal/workflows/failover.go
server/internal/workflows/pgbackrest_restore.go
server/internal/workflows/refresh_current_state.go
server/internal/workflows/restart_instance.go
server/internal/workflows/service.go
server/internal/workflows/start_instance.go
server/internal/workflows/stop_instance.go
server/internal/workflows/switchover.go
server/internal/workflows/update_database.go

🧰 Additional context used

📓 Path-based instructions (5)

server/**/*.go