Task Scheduler System

Cloudillo’s Task Scheduler is a sophisticated persistent task execution system that enables reliable, async background processing with dependency management, automatic retries, and cron-style scheduling. This system is critical for federation, file processing, and any operations that need to survive server restarts.

Why Task-Based Processing?

Simple async handlers cannot provide the guarantees needed for federation and file processing:

Problems with simple async:

✗ Lost on server restart (no persistence)
✗ No dependency ordering (file must finish before action)
✗ No automatic retry on transient failures
✗ No progress tracking or observability
✗ No priority management for resource allocation

Task scheduler solutions:

✅ Tasks survive server restarts (persisted via MetaAdapter)
✅ Dependency resolution (DAG-based ordering)
✅ Automatic retry with exponential backoff
✅ Task lifecycle tracking (pending → running → complete/failed)
✅ Priority-based execution via worker pool

Architecture Overview

Components

┌────────────────────────────────────────────┐
│ Task Scheduler                             │
│  ┌──────────────────────────────────────┐  │
│  │ Task Registry                        │  │
│  │  - TaskBuilder functions             │  │
│  │  - Task type mapping                 │  │
│  └──────────────────────────────────────┘  │
│  ┌──────────────────────────────────────┐  │
│  │ Task Queue                           │  │
│  │  - Pending tasks                     │  │
│  │  - Running tasks                     │  │
│  └──────────────────────────────────────┘  │
│  ┌──────────────────────────────────────┐  │
│  │ Dependency Tracker                   │  │
│  │  - DAG (Directed Acyclic Graph)      │  │
│  │  - Waiting tasks                     │  │
│  │  - Completion notifications          │  │
│  └──────────────────────────────────────┘  │
│  ┌──────────────────────────────────────┐  │
│  │ Retry Manager                        │  │
│  │  - Exponential backoff               │  │
│  │  - Retry counters                    │  │
│  │  - Backoff timers                    │  │
│  └──────────────────────────────────────┘  │
│  ┌──────────────────────────────────────┐  │
│  │ Cron Scheduler                       │  │
│  │  - Cron expressions                  │  │
│  │  - Next execution time               │  │
│  │  - Recurring task tracking           │  │
│  └──────────────────────────────────────┘  │
└────────────────────────────────────────────┘
                    ↓
┌────────────────────────────────────────────┐
│ MetaAdapter (Persistence)                  │
│  - Task metadata storage                   │
│  - Task status tracking                    │
│  - Dependency relationships                │
└────────────────────────────────────────────┘
                    ↓
┌────────────────────────────────────────────┐
│ Worker Pool (Execution)                    │
│  - High priority workers                   │
│  - Medium priority workers                 │
│  - Low priority workers                    │
└────────────────────────────────────────────┘

Task Trait System

All tasks implement the Task<S> trait:

#[async_trait]
pub trait Task<S: Clone>: Send + Sync + Debug {
    /// Task type identifier (e.g., "ActionCreator", "ImageResizer")
    fn kind() -> &'static str where Self: Sized;

    /// Build task from serialized context
    fn build(id: TaskId, context: &str) -> ClResult<Arc<dyn Task<S>>>
    where Self: Sized;

    /// Serialize task for persistence
    fn serialize(&self) -> String;

    /// Execute the task
    async fn run(&self, state: &S) -> ClResult<()>;

    /// Task type of this instance
    fn kind_of(&self) -> &'static str;

    /// Called once the task is permanently failed (retries exhausted, or
    /// the first failure if no retry policy). For irreversible cleanup.
    /// Default: no-op.
    async fn on_failed(&self, state: &S, attempts: u16, last_error: &str) {}
}

Retries, dependencies, keys, and cron schedules are not trait methods — they are configured at scheduling time through the builder API (see below).

Generic State Parameter

The S parameter is the application state (typically Arc<AppState>):

// Task receives app state for accessing adapters
async fn run(&self, state: &Arc<AppState>) -> ClResult<()> {
    // Access adapters through state
    state.meta_adapter.create_action(...).await?;
    state.blob_adapter.create_blob(...).await?;
    Ok(())
}

Task Serialization

Tasks must serialize to strings for persistence:

MyTask.serialize():
    return JSON.stringify(self)

MyTask.build(id, context):
    task = JSON.parse(context)
    return Arc(task)

Key Features

1. Dependency Tracking

Tasks can depend on other tasks forming a Directed Acyclic Graph (DAG):

# Create file processing task
file_task_id = scheduler
    .task(FileIdGeneratorTask(temp_path))
    .schedule()

# Create image resizing task that depends on file task
resize_task_id = scheduler
    .task(ImageResizerTask(file_id, "hd"))
    .depend_on([file_task_id])    # Wait for file task
    .schedule()

# Create action that depends on both
action_task_id = scheduler
    .task(ActionCreatorTask(action_data))
    .depend_on([file_task_id, resize_task_id])  # Wait for both
    .schedule()

Dependency Resolution:

FileIdGeneratorTask (no dependencies)
  ↓ completes
ImageResizerTask (depends on file task)
  ↓ completes
ActionCreatorTask (depends on both)
  ↓ executes

2. Exponential Backoff Retry

Tasks can retry on failure with increasing delays. RetryPolicy::new((min, max), times) takes a min/max backoff in seconds and a retry count; the default is ((60, 3600), 10). Retries only happen when the error is retryable (Error::is_retryable()) — permanent errors fail immediately. Backoff is min * 2^attempt, capped at max.

retry_policy = RetryPolicy::new((60, 3600), 5)   // 60s..1h, up to 5 retries

scheduler
    .task(ActionDeliveryTask(action_token, recipient))
    .with_retry(retry_policy)
    .schedule()

.with_automatic_retry() is a shortcut that applies the default policy.

Backoff for min=60s:

attempt 0 → 60s, attempt 1 → 120s, attempt 2 → 240s ... capped at 3600s

After times retryable failures the task is marked failed and on_failed runs.

3. Cron Scheduling

Recurring tasks use cron expressions (5-field: minute hour day month weekday) or the convenience helpers daily_at(hour, minute) and weekly_at(weekday, hour, minute) (weekday 0=Sunday). After each run the scheduler computes the next execution and re-queues the task.

# Raw cron expression — every day at 9:00 AM
scheduler.task(CleanupTask(temp_dir)).cron("0 9 * * *").schedule()

# Every day at 2:30 AM
scheduler.task(CleanupTask(temp_dir)).daily_at(2, 30).schedule()

# Every Monday at 9:00 AM
scheduler.task(WeeklyReportTask()).weekly_at(1, 9, 0).schedule()

.run_on_startup() makes a cron task fire once immediately on startup if a scheduled run was missed while the server was down (or on first registration).

4. Persistence

Tasks are persisted to MetaAdapter and survive server restarts:

start_scheduler(app):
    scheduler = Scheduler(app)

    # Load unfinished tasks from database
    pending_tasks = app.meta_adapter.list_pending_tasks()

    for task_meta in pending_tasks:
        # Rebuild task from serialized context
        task = TaskRegistry.build(
            task_meta.kind,
            task_meta.id,
            task_meta.context
        )

        # Re-queue task
        scheduler.enqueue(task_meta.id, task)

    # Start processing
    scheduler.start()

Built-in Task Types

These are the task types registered at startup (scheduler.register::<T>()):

Task	Purpose	Location
ActionCreatorTask	Creates and signs action tokens for federation	`cloudillo-action/src/task.rs`
ActionVerifierTask	Validates incoming federated action tokens	`cloudillo-action/src/task.rs`
DraftPublishTask	Publishes scheduled draft actions	`cloudillo-action/src/task.rs`
ActionDeliveryTask	Delivers actions to remote instances with retry (POST to `/api/inbox`)	`cloudillo-action/src/delivery.rs`
HistoryFetchTask	Fetches action history during federation sync	`cloudillo-action/src/history_sync.rs`
StatEmitTask	Emits aggregated stat actions (native hook)	`cloudillo-action/src/native_hooks/stat_emit_task.rs`
FileIdGeneratorTask	Computes content-addressed file IDs, moves files to BlobAdapter	`cloudillo-file/src/descriptor.rs`
ImageResizerTask	Generates image variants using Lanczos3 filter	`cloudillo-file/src/image.rs`
VideoTranscoderTask	Transcodes video to web formats	`cloudillo-file/src/video.rs`
AudioExtractorTask	Extracts audio metadata and previews	`cloudillo-file/src/audio.rs`
PdfProcessorTask	Extracts text, metadata, and page thumbnails from PDFs	`cloudillo-file/src/pdf.rs`
GcTask	Garbage-collects unreferenced blobs	`cloudillo-file/src/gc.rs`
EmailSenderTask	Sends emails asynchronously with retry	`cloudillo-email/src/task.rs`
CertRenewalTask / AcmeEarlyRetryTask	TLS certificate renewal via ACME	`cloudillo-core/src/acme.rs`
AuthCleanupTask	Cleans up expired auth/session data	`cloudillo-auth/src/cleanup.rs`
TenantImageUpdaterTask	Processes tenant avatar/banner images	`cloudillo-profile/src/media.rs`
ProfileRefreshBatchTask / ProfilePicSyncTask	Refreshes cached remote profile data and pictures	`cloudillo-profile/src/sync.rs`

Builder Pattern API

The scheduler uses a fluent builder API for task configuration:

scheduler
    .task(ActionDeliveryTask(token, recipient))
    .key(format("deliver-{}-{}", action_id, recipient))
    .depend_on(vec![action_creator_task])
    .with_retry(RetryPolicy::new((60, 3600), 5))
    .schedule()

Key methods: key() (dedup identifier), depend_on(Vec) / depends_on(single), with_retry() / with_automatic_retry(), cron() / daily_at() / weekly_at(), schedule_after(secs) / schedule_at(ts), and the terminal methods schedule(), now(), after(secs), at(ts), after_task(id).

When a key is set, scheduling deduplicates: an existing task with the same key and identical serialized parameters is reused; if parameters changed, the stored task is updated in place. There is no per-task priority on the builder — the worker pool priority (high/medium/low) is chosen inside each task’s run() when it offloads CPU work.

Integration with Worker Pool

CPU-intensive tasks should use the worker pool:

impl Task<App> for ImageResizerTask:
    run(state):
        # Load image (async I/O)
        image_data = state.blob_adapter.read_blob_buf(
            self.tn_id, self.source_file_id
        )

        # Resize image (CPU-intensive - use worker pool)
        resized = state.worker.run(closure:
            img = image.load_from_memory(image_data)
            resized = img.resize(
                self.target_width,
                self.target_height,
                FilterType::Lanczos3
            )
            # Encode to format
            buffer = resized.encode_to(self.format)
            return buffer
        )

        # Store variant (async I/O)
        variant_id = compute_file_id(resized)
        state.blob_adapter.create_blob_buf(
            self.tn_id, variant_id, resized
        )