Autoscaling

Autoscaling

Millipede uses one dispatch actor. At the beginning of each dispatch pass, it reads the desired concurrency and starts work only while in-flight attempts remain below that target. Retry backoff, host-politeness waits, and global start-budget waits stay in the dispatcher's deferred heap; deferred work does not consume an in-flight slot.

Fixed and dynamic concurrency

The crawler builder begins with fixed concurrency of 10. Calling max_concurrency(n) pins the crawler to n. Calling autoscale_mode(...) afterward clears that fixed pin while retaining n as the dynamic ceiling. Reversing the order matters: a later max_concurrency call pins the crawler again.

Fixed concurrency is useful for hard container budgets, paid-per-minute proxy plans, reproducible benchmarks, and debugging runs. A fixed value is normalized to at least one, becomes the desired, minimum, and maximum concurrency, ignores the configured scaling mode, and does not start the snapshotter or system-status loop.

AIMD convergence

Once dynamic autoscaling is enabled, additive-increase/multiplicative-decrease (AIMD) is the default. Successful attempts extend a streak. After increase_after_successes successes, desired concurrency increases by one. A failed or retryable attempt clears the streak and changes the target to round(desired * decrease_factor).

The success threshold is normalized to at least one. A decrease factor must be in (0, 1]; an invalid value falls back to 0.5. Every change is clamped to the normalized minimum and maximum concurrency.

For a live sampler that polls the crawler handle while a mock target fails transiently, see the autoscale demo. The example is intentionally not embedded here so the convergence walkthrough remains in one place.

Load-signal mode

In load-signal mode, the pool evaluates registered signals every autoscale_interval over snapshotter.window.

• If any signal's latest snapshot is overloaded, concurrency falls by ceil(current * scale_down_step_ratio), with a minimum change of one.
• Otherwise, signals with at least system_status.min_samples.max(1) observations contribute their healthy-sample ratios. If the mean reaches desired_utilization_ratio, concurrency rises by ceil(current * scale_up_step_ratio), again by at least one.
• Insufficient history or a healthy mean below the utilization target holds concurrency steady.

Built-in signals cover aggregate system CPU, used-memory ratio, Tokio timer scheduling lag, and downstream client health. ClientLoadSignal::instrument_storage(storage) wraps storage so successful operations are recorded as healthy and storage rate-limit errors as overloaded. For another downstream client, obtain ClientLoadSignal::handle() and call record_healthy() or record_rate_limited(). Selecting load-signal mode with no signals emits a warning and falls back to AIMD with a 10-success increase threshold and a 0.5 decrease factor.

Each signal decides whether a snapshot is overloaded when it records that snapshot; system status consumes the resulting boolean. The Tokio signal does not require tokio_unstable.

Bounds and controls

The pool defaults and builder behavior expose these important controls:

The task timeout is one attempt-wide deadline, not a fresh timeout for each stage. During handler work, the engine uses the earlier of this deadline and the request-handler timeout. Cleanup and post-success work are outside it.

Observing and debugging convergence

Crawler::autoscaler_snapshot() and CrawlerHandle::autoscaler_snapshot() expose desired, minimum, and maximum concurrency plus whether the target is fixed. The handle returns no snapshot after the crawler is gone. When the dispatcher observes a changed target, it emits a debug event on the millipede::concurrency tracing target with desired, current, minimum, and maximum values.

Use RUST_LOG=millipede_core=debug for crate-wide diagnostics or RUST_LOG=millipede::concurrency=debug for the concurrency target. Snapshots and tracing are the shipped observation paths and do not require a metrics exporter.

Common runaway or stalled patterns have direct checks:

Rate limiting & politeness

Per-domain delay and retry pressure

same_domain_delay spaces start reservations independently by URL host. A 429 adds an exponential per-host penalty with a one-second minimum base and a five-minute cap per penalty; a later non-429 HTTP response clears that penalty. A retry error carrying a Retry-After duration can extend the same host's next-allowed time.

The HTTP layer accepts Retry-After as delta seconds or an HTTP date and caps the trusted parsed duration at ten minutes. An external policy can provide a persistent host floor with AutoscaledPool::set_domain_delay_floor.

Politeness-delayed requests are parked by the dispatcher, so they do not occupy concurrency slots. Deferred leases are bounded to max(max_concurrency, 32). Once that buffer is full, the current queue order can still leave later hosts behind a flood from one host.

Global caps and jitter

max_tasks_per_minute creates a global token bucket for actual starts, including retries. Its capacity is at least one, it starts with one token, and it refills at capacity / 60 tokens per second. Because the dispatcher checks the bucket immediately before starting work, deferred entries do not consume tokens.

Millipede's built-in knobs provide the domain delay and global cap. Random jitter in the example below is implemented in a pre-navigation hook and is example-level policy, not a built-in limiter feature.

//! Demonstrates per-domain delay, deterministic random jitter, and crawl concurrency/rate caps.
//!
//! Run with: `cargo run -p millipede --example rate_limit`

use std::{
    hash::{DefaultHasher, Hash, Hasher},
    sync::{Arc, Mutex},
    time::{Duration, Instant},
};

use millipede::{Crawler, HttpContext, HttpKind, MemoryStorageClient};
use wiremock::{Mock, MockServer, ResponseTemplate, matchers::any};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let server = MockServer::start().await;
    Mock::given(any())
        .respond_with(ResponseTemplate::new(200).set_body_string("rate-limited page"))
        .mount(&server)
        .await;

    let entries = Arc::new(Mutex::new(Vec::<Instant>::new()));
    let kind = HttpKind::builder()
        .pre_navigation_hook(|ctx| {
            let mut hasher = DefaultHasher::new();
            ctx.request.url.as_str().hash(&mut hasher);
            let jitter_ms = hasher.finish() % 151;
            // This deterministic random jitter is example-layer behavior, not a library feature.
            Box::pin(async move {
                tokio::time::sleep(Duration::from_millis(jitter_ms)).await;
                Ok(())
            })
        })
        .build()?;
    let crawler = Crawler::builder(kind)
        .max_concurrency(2)
        .same_domain_delay(Duration::from_millis(200))
        .max_tasks_per_minute(600)
        .storage_client(Arc::new(MemoryStorageClient::new()))
        .request_handler({
            let entries = Arc::clone(&entries);
            move |ctx: HttpContext| {
                let entries = Arc::clone(&entries);
                async move {
                    // Entry is recorded first so the measurements exclude handler work.
                    entries
                        .lock()
                        .expect("handler-entry mutex poisoned")
                        .push(Instant::now());
                    println!("handled {}", ctx.request.url);
                    Ok(())
                }
            }
        })
        .build()
        .await?;

    let base = server.uri();
    let urls = (0..10)
        .map(|index| format!("{base}/page/{index}"))
        .collect::<Vec<_>>();
    let started = Instant::now();
    let stats = crawler.run(urls).await?;
    let wall_time = started.elapsed();

    let mut entries = entries
        .lock()
        .expect("handler-entry mutex poisoned")
        .clone();
    entries.sort_unstable();
    let spacings = entries
        .windows(2)
        .map(|pair| pair[1].duration_since(pair[0]))
        .collect::<Vec<_>>();
    println!("observed handler-entry spacing: {spacings:?}");
    println!("total wall time: {wall_time:?}");

    anyhow::ensure!(
        stats.requests_finished == 10 && stats.requests_failed == 0,
        "expected all ten pages to succeed, got {stats:#?}"
    );
    anyhow::ensure!(entries.len() == 10, "expected ten handler-entry samples");
    // Nine 200 ms same-domain slots imply about 1.8 s. The 1.4 s bound leaves broad scheduler
    // slack while still failing if the domain delay is not applied at all.
    anyhow::ensure!(
        wall_time >= Duration::from_millis(1_400),
        "crawl completed too quickly for the configured same-domain delay: {wall_time:?}"
    );
    Ok(())
}

Run it with cargo run -p millipede --example rate_limit.

Next steps

• Follow AIMD changes in the autoscale demo.
• Combine politeness with identity controls in sessions and proxies.