Fleet

Introduction

Featurefleet Fleet runs your queue workers and shows you what they are doing. One command, navree fleet, starts a supervisor that keeps the right number of worker processes on each queue, moves workers to the queues that need them, restarts workers that exit and recycles them before they grow. A dashboard at /_fleet, served by your app, shows each queue's throughput, wait time and runtimes, every failed job with a button to retry or forget it, and controls to pause and continue the workers.

You describe your workers once, in configuration, as supervisors: which queues they work, how many processes they may use and how they balance. Every server runs the same navree fleet command with the same configuration, and the dashboard shows all of them together.

Fleet builds on navree-queue's database driver. Its workers run jobs with the same code as queue:work, so jobs, retries, backoff, time limits and failed jobs behave exactly as the queues page describes.

Installation

The fleet feature of the navree facade brings the crate navree-fleet. It is part of full, the default, and needs no system library. Add the module after QueueModule:

pub fn app(sources: ConfigSources) -> AppBuilder {
    Navree::app()
        .module(ConfigModule::for_root(sources))
        .module(HttpModule::from_config())
        .module(DatabaseModule::from_config())
        .module(QueueModule::from_config())
        .module(FleetModule::from_config())
        // ...
}

FleetModule is global and provides the Fleet service. It needs DatabaseModule and QueueModule; the fleet command refuses the memory queue driver, whose jobs only exist in one process. Its dashboard routes declare access, so with the dashboard on (the default) the app also needs an authenticator (AuthModule) and navree-access's route authorizer (AccessModule); set fleet.dashboard = false in an app without them. With EventsModule in the app, a running master emits long-wait events. With ObserveModule, GET /metrics includes Fleet's gauges.

The migration set fleet (after queue) creates four tables: fleet_masters (the running masters, their supervisors and processes, and their heartbeats), fleet_state (paused supervisors, the last fleet:terminate and long-wait cooldowns), fleet_jobs (recently finished job attempts) and fleet_metrics (snapshots per queue and job type). On TimescaleDB the last two are hypertables on their Unix millisecond columns, an hour and a day per chunk, and trimming drops whole chunks.

navree migrate

Supervisors

A supervisor is a group of worker processes with one configuration. Declare them in the fleet.supervisors table, one entry per supervisor:

[fleet.supervisors.default]
queues = ["default", "mail", "notifications"]
balance = "auto"
min_processes = 1     # per queue
max_processes = 8     # for the whole supervisor
max_jobs = 1000       # recycle a worker after this many jobs
max_memory_mb = 256   # ... or when it uses more memory than this

[fleet.supervisors.exports]
queues = ["exports"]
balance = "off"
max_processes = 2

Then start Fleet:

navree fleet

Fleet started on worker-1 (pid 48211) with 2 supervisors.
  default  default, mail, notifications: auto, 1-8 processes
  exports  exports: off, 2 processes

fleet is a command of your app's binary, so in production you run ./crm-api fleet. The process that runs it is the master: it runs no jobs itself, it watches the workers and the queues. It starts each worker as a child process of the same binary running fleet:work, a worker with its supervisor's queues and limits, and every worker runs up to queue.workers jobs at once (4), or the supervisor's concurrency when you set it. The master talks to its workers over their standard input (pause, continue, stop) and reads their reports, lines starting with @fleet, from their standard output; anything else a worker prints, such as its logs, the master prints as it is. The master's name is the server's host name.

Balancing

balance decides how a supervisor's processes are spread over its queues:

With auto, every fleet.balance_cooldown_secs (3 seconds) the master measures each queue's workload: its waiting and running jobs times the average runtime of its jobs over the last five minutes (1 second per job when none ran lately), the time the queue would take to clear on one process. It gives each queue min_processes, shares the rest of max_processes between the queues in proportion to their workload, never giving a queue more processes than it can keep busy (its waiting and running jobs over the jobs a process runs at once), and moves each queue towards that share by at most fleet.balance_max_shift processes (1) per step, so the supervisor changes shape steadily instead of jumping on every spike. Delayed jobs count once they are due.

Take the default supervisor above at 09:00, when a newsletter queues 3,000 mails. mail grows from one process to six over the next 15 seconds while default and notifications keep one each. As mail drains, its processes return to one, and the spare capacity goes to whichever queue fills up next. A queue that is empty keeps min_processes, so the first job on it starts at once.

min_processes times the number of queues must not exceed max_processes, and a simple supervisor needs at least one process per queue; a supervisor that breaks a rule fails the boot with its name.

The balancer is a pure function, navree::fleet::next_allocation(&supervisor, &current, &workload, max_shift), next to target_allocation (where the supervisor would end up) and initial_allocation, so you can check a supervisor's configuration against any workload in a test.

Recycling and restarts

Long-running processes accumulate state. Fleet replaces a worker when it reaches one of its supervisor's limits:

A worker that reaches a limit takes no new job, finishes the ones it is running and exits, and the master starts a fresh one in its place at once. A worker that exits unexpectedly (a crash, a kill -9) or cannot be started is restarted after a delay that grows from 1 second to 60 seconds while it keeps failing; five failures within a minute are logged as an error under navree::fleet. A job that a dead worker had reserved runs again once its reservation expires, as on the queues page.

Stopping

Ctrl-C, or SIGTERM on Unix, stops Fleet gracefully: the master stops starting workers, tells each worker to stop, waits while they finish the jobs they are running, and exits. A worker whose master goes away (its standard input closes) stops the same way. A job is never cut off by the shutdown, only by its own timeout_secs, so give your process manager a stop timeout longer than your longest job's limit.

Running Fleet in production

Run one fleet process per worker server, under a process manager, in place of the queue:work processes you would otherwise run. With Harbor, make the worker process run fleet:

[processes]
worker = { command = "fleet" }

Without Harbor, a systemd unit is enough:

[Unit]
Description=CRM queue workers (Fleet)
After=network-online.target
Wants=network-online.target

[Service]
User=crm
WorkingDirectory=/srv/crm-api/current
EnvironmentFile=/etc/crm-api/env
ExecStart=/srv/crm-api/current/crm-api fleet
Restart=always
# Workers finish their jobs before Fleet exits: longer than your longest job.
TimeoutStopSec=180
KillMode=mixed

[Install]
WantedBy=multi-user.target

KillMode=mixed sends the stop signal to the master only, which stops its workers itself.

• After every deploy, restart Fleet so its workers run the new code. Restart the unit, or run navree fleet:terminate: every master on every server finishes its jobs and exits, and the process manager starts it again from the new release. A master started after the fleet:terminate keeps running.
• Several servers. Every server runs fleet with the same configuration. Each master balances its own processes; the dashboard and fleet:status show them all. To give one server more processes, override the limit in its environment: APP_FLEET__SUPERVISORS__DEFAULT__MAX_PROCESSES=16.
• Dedicated servers. Give a server only some supervisors with fleet.only: APP_FLEET__ONLY=exports on a large machine that runs nothing but exports.

Pausing and continuing

Pause a supervisor to stop it taking jobs, for example while a provider your jobs call is down. Its workers finish what they are running and then wait; jobs keep queueing, nothing is lost.

navree fleet:pause exports    # one supervisor
navree fleet:pause            # every supervisor, on every server
navree fleet:continue exports
navree fleet:continue

The state is stored in the database, so a pause applies to every server at once (each master checks every second) and survives restarts and deploys. The dashboard has the same buttons, and the Fleet service the same methods:

use navree::fleet::Fleet;

fleet.pause(Some("exports")).await?;
fleet.resume(Some("exports")).await?;
let paused = fleet.paused().await?; // the paused supervisors

The dashboard

The dashboard is a page served by your app at fleet.path (/_fleet), outside the API prefix: HTML rendered with navree-view, with its script and styles compiled into the crate and served from /_fleet/assets/. It reads the JSON API under /_fleet/api and refreshes every 5 seconds.

• Overview. A row of figures: Fleet's status (running, paused when every supervisor is paused, or inactive when no master has reported for 30 seconds), jobs per minute, jobs in the past hour, failed jobs in the past seven days, and the longest current wait with the queue it is on. Below, the workload table: each queue with its pending jobs, its processes, the wait of its oldest job, the time it would take to clear and its jobs per minute.
• Supervisors. Every master, by server, with its supervisors, a button to pause or continue each, and their processes. Each process shows its queue, its pid, its memory, the jobs it has run and when it started.
• Jobs. Pending jobs (waiting, delayed and running) and job attempts finished within fleet.trim.recent_minutes (the last hour), with their type, queue, outcome, runtime and when they ran. Search by job type or queue, and open a pending job to see its payload and attempts.
• Failed jobs. Every job in failed_jobs, newest first, with its type, queue, attempts, when it failed and its error. Search by type, queue, error or payload text. Open one to read the full error and the payload, then Retry it or Forget it. Retry all retries every failed job matching the search, so one click puts back every contacts.export that failed while the mail server was down.
• Metrics. For each queue and each job type, charts of throughput and of average and p95 runtime, and for each queue its wait time, over the last hour (per minute), day (per 15 minutes) or week (per hour).

Payload values whose names contain a fragment of log.redact (password, token, secret and the rest) are shown as [redacted], in the page and in the API.

Access

The page and its assets are public and hold no data. The page signs in through your app's sign-in route, navree-auth's POST {http.api_prefix}{auth.routes_prefix}/login (/api/auth/login, two-factor codes included), keeps the access and refresh tokens in the browser tab's session storage, and sends Authorization: Bearer to the API; you can also paste an access token. It is served with a strict Content Security Policy (no inline scripts) and X-Frame-Options: DENY. Fleet registers the subject Fleet with the permission prefix fleet, and every API route checks an ability on it:

navree access:grant manager fleet.view
navree access:grant admin fleet.operate

Metrics

Workers turn navree-queue's telemetry job events into finished attempts and report them to their master as they go. The master stores them in one batch per heartbeat (every 3 seconds) in fleet_jobs, kept for fleet.trim.recent_minutes (60), and every fleet.snapshot_secs (60) it stores one snapshot per queue it supervises and per job type that ran: attempts finished and failed, average and p95 runtime, and each queue's pending count and oldest wait. Snapshots are kept for fleet.trim.metrics_days (7).

With ObserveModule in the app, GET /metrics on your web server includes Fleet's latest numbers, read from the database when Prometheus scrapes, so you don't have to scrape the worker servers:

Long-wait alerts

A queue that falls behind shows up as wait time before it shows up anywhere else. Set a threshold per queue, in seconds:

[fleet.long_wait]
default = 60
mail = 30
exports = 600

When a queue's oldest available job has waited longer than its threshold, a running master emits LongWaitDetected (event name fleet.long_wait, fields queue, wait_secs and threshold_secs), at most once per queue every fleet.long_wait_cooldown_minutes (15) across every master, and logs a warning under navree::fleet. Queues without a threshold never raise one. Listen to the event to send it where your team looks:

use navree::BoxFuture;
use navree::events::Listener;
use navree::fleet::LongWaitDetected;
use navree::prelude::*;

#[injectable]
pub struct PostToOpsChannel {
    chat: Arc<ChatClient>,
}

impl Listener<LongWaitDetected> for PostToOpsChannel {
    fn handle<'a>(
        &'a self,
        event: &'a LongWaitDetected,
        _ctx: &'a AppContext,
    ) -> BoxFuture<'a, Result<(), BoxError>> {
        Box::pin(async move {
            let text = format!(
                "Queue `{}` is behind: its oldest job has waited {} s (threshold {} s).",
                event.queue, event.wait_secs, event.threshold_secs
            );
            self.chat.post("#ops", &text).await?;
            Ok(())
        })
    }
}

#[module(register = Self::wire)]
pub struct OpsModule;

impl OpsModule {
    fn wire(&self, m: &mut ModuleBuilder) {
        m.listen::<LongWaitDetected, PostToOpsChannel>();
    }
}

Fleet::long_waits() returns the queues over their threshold right now, and the dashboard marks a queue in amber while it is over its threshold.

Failed jobs

Fleet works on the same failed_jobs table as the queue commands, so queue:failed, queue:retry and queue:flush keep working next to the dashboard. Fleet adds one command, to delete a single failed job:

navree fleet:forget 0199c1d2-6f7e-7a31-b5d4-2f0c8e9a1b23

The Fleet service has the same operations: failed_jobs(search), failed_job(id), retry(id), retry_all(search) and forget(id).

Configuration

Every key of the fleet section is optional except each supervisor's queues. From the environment, nested keys use __ and lists are comma-separated strings: APP_FLEET__SUPERVISORS__DEFAULT__QUEUES=default,mail.

An app without a fleet.supervisors table gets one supervisor, default, working queue.queues (or queue.default_queue) with the defaults above, and enough processes for one per queue when queue.queues lists more than four.

Fleet is running.

Master    Status   Supervisors  Processes  Started
worker-1  running  2            7          2026-10-26 09:15:12
worker-2  running  1            2          2026-10-26 09:15:40

Queue          Pending  Oldest wait  Processes  Jobs/min
default        12       3 s          3          420
mail           0        0 s          1          35
notifications  0        0 s          1          18
exports        2        41 s         4          3

Testing

Test your jobs as you always do, with QueueFake or a real queue and run_pending. To test what Fleet decides, use the Fleet service with a fixed clock. Here a queued mail that has waited two minutes is a long wait:

use std::time::Duration;

use chrono::{TimeZone, Utc};
use navree::contracts::Clock;
use navree::fleet::Fleet;
use navree::prelude::*;
use navree::queue::Queue;
use navree::testing::{FixedClock, TestApp};
use uuid::Uuid;

#[tokio::test]
async fn a_mail_waiting_two_minutes_is_a_long_wait() {
    let clock = Arc::new(FixedClock::new(Utc.with_ymd_and_hms(2026, 10, 26, 9, 0, 0).unwrap()));
    let app = TestApp::builder()
        .temp_database()
        .config("fleet.long_wait.mail", 60)
        .config("fleet.dashboard", false) // no AccessModule in this test app
        .module(DatabaseModule::from_config())
        .module(QueueModule::from_config())
        .module(FleetModule::from_config())
        .module(DigestModule)
        .override_value::<dyn Clock>(clock.clone())
        .build()
        .await;

    // SendDigest's options put it on the `mail` queue.
    app.get::<Queue>().await.dispatch(SendDigest { user_id: Uuid::now_v7() }).await.unwrap();
    clock.advance(Duration::from_secs(120));

    let late = app.get::<Fleet>().await.long_waits().await.unwrap();
    assert_eq!(late.len(), 1);
    assert_eq!(late[0].queue, "mail");
    assert_eq!(late[0].wait_secs, 120);
    app.shutdown().await.unwrap();
}

Fleet::workload() returns each queue's pending count, oldest wait, average runtime, workload and time to clear, the numbers the balancer works from, so you can check a supervisor's configuration against a workload you build in a test with next_allocation.

To test a master itself, run Master::new(fleet) (or Master::from_app(&ctx)) with another spawner: navree::fleet::testing::FakeSpawner starts no process and lets the test read the commands each worker received and make workers report, reach a limit or crash, and InProcessSpawner runs real workers as tasks of the test app. MasterSettings shortens the master's pace for tests.