[ENH] Schedule tasks from the sysdb.

[rescrv]

Description of changes

This PR adds a scheduler that talks to the sysdb to pull tasks.

Test plan

CI

Migration plan

N/A

Observability plan

N/A

Documentation Changes

N/A

[github-actions]

Reviewer Checklist

Please leverage this checklist to ensure your code review is thorough before approving

Testing, Bugs, Errors, Logs, Documentation

• Can you think of any use case in which the code does not behave as intended? Have they been tested?
• Can you think of any inputs or external events that could break the code? Is user input validated and safe? Have they been tested?
• If appropriate, are there adequate property based tests?
• If appropriate, are there adequate unit tests?
• Should any logging, debugging, tracing information be added or removed?
• Are error messages user-friendly?
• Have all documentation changes needed been made?
• Have all non-obvious changes been commented?

System Compatibility

• Are there any potential impacts on other parts of the system or backward compatibility?
• Does this change intersect with any items on our roadmap, and if so, is there a plan for fitting them together?

Quality

• Is this code of a unexpectedly high quality (Readability, Modularity, Intuitiveness)

[propel-code-bot]

Add Heap Scheduler Integration between sysdb and Heap Tender (Task Scheduling Orchestration)

This PR introduces substantial orchestration between the Rust heap scheduling/compaction layer (s3heap-service) and the Go sysdb coordination layer to enable scheduled task execution based on sysdb state. It includes protocol additions (gRPC/protobuf), new sysdb APIs for task scheduling/advancement, and a full Rust integration that pulls, schedules, and acknowledges tasks from sysdb, wiring real scheduler state into HeapTender. Additionally, several tests, database DAO implementations, and minor refactors are included to fully close the end-to-end loop for distributed heap scheduling and durable task progression.

Key Changes

• Introduced AdvanceTask RPC in sysdb: Plumbed through Go protobuf, coordinator, DAO, and gRPC server.
• Added PeekScheduleByCollectionId RPC for batch scheduling lookups based on collection IDs.
• Expanded Go sysdb DAO (taskDb) and interface to support GetByID, AdvanceTask, and PeekScheduleByCollectionId, with new test coverage in task_test.go.
• Integrated sysdb-based scheduling and nonce advancement into the Rust HeapTender orchestration logic-now triggers sysdb peeks, converts schedule entries, and calls HeapWriter.
• Expanded Rust types to support task schedule entries (ScheduleEntry) and conversion from proto to type-safe representations.
• Wire up sysdb config and client into the Rust heap tender service configuration and startup flow; enforce correct sysdb mode (grpc, not sqlite).
• Schema/proto changes: Added/updated messages for AdvanceTask, PeekScheduleByCollectionId, and schedule entry representations.
• Refactored Rust S3heap schedule/triggerable types to be copyable, hashable, and comparable for efficient use in collections and deduplication logic.
• Numerous test updates and additions covering DAO correctness, schedule peeking/filtering, and end-to-end tendering.
• Cargo.toml/lock and workspace dependency updates for new sysdb/chrono dependencies and propagation.

Affected Areas

• go/pkg/sysdb/coordinator/task.go (protos/coordination; business logic)
• go/pkg/sysdb/metastore/db/dao/task.go and related DAO/test files
• idl/chromadb/proto/coordinator.proto (protobuf API surface)
• rust/s3heap-service/ (src/lib.rs - HeapTender logic; config; startup)
• rust/sysdb/ (sysdb-side proto bindings, error handling, and peeks)
• rust/types/ and rust/s3heap/ (task schedule and heap types, S3heap scheduler interface)
• rust/s3heap/tests/ and rust/s3heap-service/tests/ (integration and unit tests)
• Cargo.toml and Cargo.lock files (dependency tree)

This summary was automatically generated by @propel-code-bot

[propel-code-bot]

[PerformanceOptimization]

You've correctly identified a potential optimization with the comment // Can do both calls with a JOIN. To avoid making two separate database calls, you could implement a dedicated DAO method that performs a JOIN between the tasks and operators tables. This would improve efficiency and prevent potential N+1 query issues if this function is called in a loop in the future.

For example, you could add a method like GetByNameWithOperator to your ITaskDb interface:

// In dbmodel/task.go
type TaskAndOperator struct {
    Task
    OperatorName string `gorm:"column:operator_name"`
}

// In ITaskDb interface
GetByNameWithOperator(inputCollectionID string, taskName string) (*TaskAndOperator, error)

// In dao/task.go
func (s *taskDb) GetByNameWithOperator(inputCollectionID string, taskName string) (*dbmodel.TaskAndOperator, error) {
    var result dbmodel.TaskAndOperator
    err := s.db.Table("tasks").
        Select("tasks.*, operators.operator_name").
        Joins("JOIN operators ON tasks.operator_id = operators.operator_id").
        Where("tasks.input_collection_id = ?", inputCollectionID).
        Where("tasks.task_name = ?", taskName).
        Where("tasks.is_deleted = ?", false).
        First(&result).Error

    if err != nil {
        if errors.Is(err, gorm.ErrRecordNotFound) {
            return nil, nil
        }
        return nil, err
    }
    return &result, nil
}

Then GetTaskByName in the coordinator can be simplified to a single database call.

Context for Agents

[**PerformanceOptimization**]

You've correctly identified a potential optimization with the comment `// Can do both calls with a JOIN`. To avoid making two separate database calls, you could implement a dedicated DAO method that performs a JOIN between the `tasks` and `operators` tables. This would improve efficiency and prevent potential N+1 query issues if this function is called in a loop in the future.

For example, you could add a method like `GetByNameWithOperator` to your `ITaskDb` interface:

```go
// In dbmodel/task.go
type TaskAndOperator struct {
    Task
    OperatorName string `gorm:"column:operator_name"`
}

// In ITaskDb interface
GetByNameWithOperator(inputCollectionID string, taskName string) (*TaskAndOperator, error)

// In dao/task.go
func (s *taskDb) GetByNameWithOperator(inputCollectionID string, taskName string) (*dbmodel.TaskAndOperator, error) {
    var result dbmodel.TaskAndOperator
    err := s.db.Table("tasks").
        Select("tasks.*, operators.operator_name").
        Joins("JOIN operators ON tasks.operator_id = operators.operator_id").
        Where("tasks.input_collection_id = ?", inputCollectionID).
        Where("tasks.task_name = ?", taskName).
        Where("tasks.is_deleted = ?", false).
        First(&result).Error

    if err != nil {
        if errors.Is(err, gorm.ErrRecordNotFound) {
            return nil, nil
        }
        return nil, err
    }
    return &result, nil
}
```

Then `GetTaskByName` in the coordinator can be simplified to a single database call.

File: go/pkg/sysdb/coordinator/task.go
Line: 146

[rescrv]

Not for this PR.

[propel-code-bot]

[BestPractice]

The tasks table defines an operator_id column but lacks a foreign key constraint to the operators table. This could lead to data integrity issues if an operator is deleted, leaving tasks with orphaned operator_id references. This would cause GetTaskByName to fail when it tries to look up the operator.

To ensure referential integrity, please consider adding a foreign key constraint.

Suggested Change

Suggested change

	CREATE TABLE "public"."tasks" (
	"task_id" uuid NOT NULL,
	"task_name" text NOT NULL,
	"tenant_id" text NOT NULL,
	"database_id" text NOT NULL,
	"input_collection_id" text NOT NULL, -- Keeping these as text instead of UUID until collections.id becomes a UUID
	"output_collection_name" text NOT NULL,
	"operator_id" uuid NOT NULL,
	"operator_params" jsonb NOT NULL,
	"completion_offset" bigint NOT NULL DEFAULT 0,
	"last_run" timestamp NULL DEFAULT NULL,
	"next_run" timestamp NULL DEFAULT NULL,
	"min_records_for_task" bigint NOT NULL DEFAULT 100,
	"current_attempts" integer NOT NULL DEFAULT 0,
	"is_alive" boolean NOT NULL DEFAULT true,
	"is_deleted" boolean NOT NULL DEFAULT false,
	"created_at" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
	"updated_at" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
	"task_template_parent" uuid NULL,
	"next_nonce" UUID NOT NULL, -- UUIDv7
	"oldest_written_nonce" UUID DEFAULT NULL, -- UUIDv7
	PRIMARY KEY ("task_id"),
	CONSTRAINT "unique_task_per_collection" UNIQUE ("input_collection_id", "task_name")
	);
	CREATE TABLE "public"."tasks" (
	"task_id" uuid NOT NULL,
	"task_name" text NOT NULL,
	"tenant_id" text NOT NULL,
	"database_id" text NOT NULL,
	"input_collection_id" text NOT NULL, -- Keeping these as text instead of UUID until collections.id becomes a UUID
	"output_collection_name" text NOT NULL,
	"operator_id" uuid NOT NULL,
	"operator_params" jsonb NOT NULL,
	"completion_offset" bigint NOT NULL DEFAULT 0,
	"last_run" timestamp NULL DEFAULT NULL,
	"next_run" timestamp NULL DEFAULT NULL,
	"min_records_for_task" bigint NOT NULL DEFAULT 100,
	"current_attempts" integer NOT NULL DEFAULT 0,
	"is_alive" boolean NOT NULL DEFAULT true,
	"is_deleted" boolean NOT NULL DEFAULT false,
	"created_at" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
	"updated_at" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
	"task_template_parent" uuid NULL,
	"next_nonce" UUID NOT NULL, -- UUIDv7
	"oldest_written_nonce" UUID DEFAULT NULL, -- UUIDv7
	PRIMARY KEY ("task_id"),
	CONSTRAINT "unique_task_per_collection" UNIQUE ("input_collection_id", "task_name"),
	CONSTRAINT "fk_tasks_operator_id" FOREIGN KEY ("operator_id") REFERENCES "public"."operators" ("operator_id")
	);

⚡ Committable suggestion

Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.

Context for Agents

[**BestPractice**]

The `tasks` table defines an `operator_id` column but lacks a foreign key constraint to the `operators` table. This could lead to data integrity issues if an operator is deleted, leaving tasks with orphaned `operator_id` references. This would cause `GetTaskByName` to fail when it tries to look up the operator.

To ensure referential integrity, please consider adding a foreign key constraint.

<details>
<summary>Suggested Change</summary>

```suggestion
CREATE TABLE "public"."tasks" (
  "task_id" uuid NOT NULL,
  "task_name" text NOT NULL,
  "tenant_id" text NOT NULL,
  "database_id" text NOT NULL,
  "input_collection_id" text NOT NULL, -- Keeping these as text instead of UUID until collections.id becomes a UUID
  "output_collection_name" text NOT NULL,
  "operator_id" uuid NOT NULL,
  "operator_params" jsonb NOT NULL,
  "completion_offset" bigint NOT NULL DEFAULT 0,
  "last_run" timestamp NULL DEFAULT NULL,
  "next_run" timestamp NULL DEFAULT NULL,
  "min_records_for_task" bigint NOT NULL DEFAULT 100,
  "current_attempts" integer NOT NULL DEFAULT 0,
  "is_alive" boolean NOT NULL DEFAULT true,
  "is_deleted" boolean NOT NULL DEFAULT false,
  "created_at" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  "updated_at" timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
  "task_template_parent" uuid NULL,
  "next_nonce" UUID NOT NULL, -- UUIDv7
  "oldest_written_nonce" UUID DEFAULT NULL, -- UUIDv7
  PRIMARY KEY ("task_id"),
  CONSTRAINT "unique_task_per_collection" UNIQUE ("input_collection_id", "task_name"),
  CONSTRAINT "fk_tasks_operator_id" FOREIGN KEY ("operator_id") REFERENCES "public"."operators" ("operator_id")
);
```

⚡ **Committable suggestion**

Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.

</details>

File: go/pkg/sysdb/metastore/db/migrations/20251001073000.sql
Line: 43

[propel-code-bot]

[BestPractice]

The current implementation of camel_to_snake_case incorrectly handles acronyms in Go constant names. For example, a name like HTTPGetter would be converted to h_t_t_p_getter instead of the correct http_getter. While this works for the current RecordCounter constant, it will cause issues with future constants that use acronyms.

A more robust implementation should handle consecutive uppercase letters correctly.

Suggested Change

Suggested change

	fn camel_to_snake_case(s: &str) -> String {
	let mut result = String::new();
	for (i, ch) in s.chars().enumerate() {
	if ch.is_uppercase() {
	if i > 0 {
	result.push('_');
	}
	result.push(ch.to_ascii_lowercase());
	} else {
	result.push(ch);
	}
	}
	result
	}
	fn camel_to_snake_case(s: &str) -> String {
	let mut result = String::new();
	let mut last_was_upper = false;
	for (i, ch) in s.chars().enumerate() {
	if ch.is_uppercase() {
	if i > 0 && !last_was_upper {
	result.push('_');
	}
	result.push(ch.to_ascii_lowercase());
	last_was_upper = true;
	} else {
	result.push(ch);
	last_was_upper = false;
	}
	}
	result
	}

⚡ Committable suggestion

Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.

Context for Agents

[**BestPractice**]

The current implementation of `camel_to_snake_case` incorrectly handles acronyms in Go constant names. For example, a name like `HTTPGetter` would be converted to `h_t_t_p_getter` instead of the correct `http_getter`. While this works for the current `RecordCounter` constant, it will cause issues with future constants that use acronyms.

A more robust implementation should handle consecutive uppercase letters correctly.
<details>
<summary>Suggested Change</summary>

```suggestion
fn camel_to_snake_case(s: &str) -> String {
    let mut result = String::new();
    let mut last_was_upper = false;
    for (i, ch) in s.chars().enumerate() {
        if ch.is_uppercase() {
            if i > 0 && !last_was_upper {
                result.push('_');
            }
            result.push(ch.to_ascii_lowercase());
            last_was_upper = true;
        } else {
            result.push(ch);
            last_was_upper = false;
        }
    }
    result
}
```

⚡ **Committable suggestion**

Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.

</details>

File: rust/types/operator_codegen.rs
Line: 148

[propel-code-bot]

[BestPractice]

Using panic! for configuration errors can cause the service to crash abruptly on startup, which can be difficult to debug in production environments. Since this function already returns a Result, it would be more robust to return an Err variant for invalid configurations. This allows for graceful error handling and clearer error messages upon service startup failure.

Context for Agents

[**BestPractice**]

Using `panic!` for configuration errors can cause the service to crash abruptly on startup, which can be difficult to debug in production environments. Since this function already returns a `Result`, it would be more robust to return an `Err` variant for invalid configurations. This allows for graceful error handling and clearer error messages upon service startup failure.

File: rust/s3heap-service/src/lib.rs
Line: 282

[tanujnay112]

Can we put a verb in this method's name to make it clear that this modifies state?

[propel-code-bot]

[BestPractice]

Resource leak: The test creates and inserts a task but uses Unscoped().Delete() for cleanup, which bypasses GORM hooks and constraints. If the test fails before reaching cleanup, the task will remain in the database permanently. Use proper cleanup with defer or t.Cleanup():

func (suite *TaskDbTestSuite) TestTaskDb_GetByID() {
    // ... setup code ...
    
    // Clean up immediately after insert, regardless of test outcome
    suite.T().Cleanup(func() {
        suite.db.Unscoped().Delete(&dbmodel.Task{}, "task_id = ?", task.ID)
    })
    
    err := suite.Db.Insert(task)
    suite.Require().NoError(err)
    // ... rest of test ...
}

Context for Agents

[**BestPractice**]

Resource leak: The test creates and inserts a task but uses `Unscoped().Delete()` for cleanup, which bypasses GORM hooks and constraints. If the test fails before reaching cleanup, the task will remain in the database permanently. Use proper cleanup with `defer` or `t.Cleanup()`:

```go
func (suite *TaskDbTestSuite) TestTaskDb_GetByID() {
    // ... setup code ...
    
    // Clean up immediately after insert, regardless of test outcome
    suite.T().Cleanup(func() {
        suite.db.Unscoped().Delete(&dbmodel.Task{}, "task_id = ?", task.ID)
    })
    
    err := suite.Db.Insert(task)
    suite.Require().NoError(err)
    // ... rest of test ...
}
```

File: go/pkg/sysdb/metastore/db/dao/task_test.go
Line: 343

[tanujnay112]

Should we consider generating the next nonce here instead of letting the caller supply it?

[rescrv]

Is that not what the code is doing?

[tanujnay112]

Naming inconsistency with FinishTask

[rescrv]

Updated.

[tanujnay112]

ditto

[rescrv]

Ditto.

[propel-code-bot]

[BestPractice]

Missing error handling: If self.sysdb.peek_schedule_by_collection_id() fails, the method returns early with ? but doesn't log the failure or provide context about which collections caused the issue. This makes debugging production failures difficult. Add error logging:

let scheduled = self
    .sysdb
    .clone()
    .peek_schedule_by_collection_id(&collection_ids)
    .await
    .map_err(|e| {
        log::error!("Failed to peek schedule for collections {:?}: {}", collection_ids, e);
        e
    })?;

Context for Agents

[**BestPractice**]

Missing error handling: If `self.sysdb.peek_schedule_by_collection_id()` fails, the method returns early with `?` but doesn't log the failure or provide context about which collections caused the issue. This makes debugging production failures difficult. Add error logging:

```rust
let scheduled = self
    .sysdb
    .clone()
    .peek_schedule_by_collection_id(&collection_ids)
    .await
    .map_err(|e| {
        log::error!("Failed to peek schedule for collections {:?}: {}", collection_ids, e);
        e
    })?;
```

File: rust/s3heap-service/src/lib.rs
Line: 54

[tanujnay112]

FYI I'm going to remove/modify this later. The path that updates next_nonce also needs to transactionally update completion_offset.