sort-governor 0.1.3

Process-wide governor for bounded asynchronous sorting: plans in-memory versus external sorts and rations file descriptors and spill memory
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

# sort-governor

[![Crates.io][crates-badge]][crates-url]
[![Documentation][docs-badge]][docs-url]
[![CI][ci-badge]][ci-url]
[![License][license-badge]][license-url]
[![Downloads][downloads-badge]][downloads-url]

A process-wide governor for bounded asynchronous sorting.

Every sort in a process — small or huge, in memory or spilling to disk —
is admitted by one `SorterHandle`. The governor decides **in-memory versus
external** for each sort from its estimated size and the *live* memory and
file-descriptor pressure, rations the **process-global** sort resources
(open descriptors, spill memory, concurrent external sorts), and runs a
**bounded, fully asynchronous cascade merge** whose memory and descriptor
footprint stay `O(fan_in)` for any input size.

## Why a governor?

Constructing an external sort ad hoc at every call site produces four
independent failures once sorts run concurrently or inputs get large:

1. **Unbounded merge fan-in.** Opening one reader per spilled run exhausts
   the process descriptor table (`EMFILE`) as soon as run count exceeds the
   soft limit.
2. **Two budgets that fight.** A memory limit that shrinks the run buffer
   inflates the run count — and therefore the descriptor count. Only a
   component that sees *both* budgets can size a run buffer *against* the
   fan-in instead of letting one knob sabotage the other.
3. **Blocking I/O on the async runtime.** A synchronous reader driven from
   an async stream stalls the executor's worker threads.
4. **No global admission.** Even with every sort individually bounded, `N`
   concurrent sorts collectively exhaust the process.

`sort-governor` addresses all four: one pure planner reconciles memory and
descriptor pressure, one actor rations permits, every run reader holds
exactly one descriptor and reads one row at a time over
[`async-fs-io`](https://crates.io/crates/async-fs-io), and the cascade
merge opens at most `fan_in` readers plus one output writer. During ingestion,
a 64-level frontier combines adjacent runs as soon as their levels match;
spill-path metadata stays fixed even when the input estimate is too small.

## Usage

```rust
use std::sync::Arc;

use futures_util::StreamExt;
use sort_governor::{MemoryPressure, SortSpec, SorterConfig, SorterError, SorterHandle, StaticPressure};

#[tokio::main]
async fn main() -> Result<(), SorterError> {
    // Wire in your process memory governor here; a fixed reading works for
    // processes without one.
    let pressure: Arc<dyn MemoryPressure> = Arc::new(StaticPressure::new(1 << 30, 0));
    let scratch_root = std::env::temp_dir().join("sort-governor-example");

    // 256 usable descriptors, of which the governor may ration 160.
    let sorter = SorterHandle::spawn(SorterConfig::from_fd_limit(256), 160, pressure, scratch_root);

    // Describe the sort; the governor plans it and hands back a lease that
    // holds the sort's resource permits.
    let lease = sorter.submit(SortSpec::new(5, 5 * 16).labelled("example")).await?;
    let mut session = lease.into_session::<u32, String>(false);
    for key in [5_u32, 1, 4, 2, 3] {
        session.push(key, format!("row {key}")).await?;
    }

    // Values stream out in key order; the lease and any scratch directory are
    // released when the stream is fully consumed or dropped.
    let mut stream = session.finish().await?;
    let mut ordered = Vec::new();
    while let Some(value) = stream.next().await {
        ordered.push(value?);
    }
    assert_eq!(ordered[0], "row 1");
    assert_eq!(ordered[4], "row 5");
    Ok(())
}
```

## How it decides

| Type              | Role                                                                                                   |
| ----------------- | ------------------------------------------------------------------------------------------------------ |
| `SortSpec`        | The caller's estimate: rows, bytes, dedup, label. The planner never sees the rows themselves.          |
| `SorterConfig`    | Process-lifetime caps: in-memory ceiling, max fan-in, concurrent external sorts, run-buffer bounds.    |
| `MemoryPressure`  | Live memory readings (effective target, resident bytes). Implement it over your memory governor.       |
| `SorterSnapshot`  | One point-in-time view of memory *and* descriptor pressure — the input the planner reconciles.         |
| `SortPlanner`     | Pure and stateless: `(spec, snapshot, config) → SortPlan`. Exhaustively unit-testable.                 |
| `SortPlan`        | `InMemory`, or `External { run_buffer_bytes, max_fan_in }`.                                            |
| `SorterHandle`    | Cloneable client of the one governor actor; `submit` returns a lease, `stats` reports counters.        |
| `SortLease`       | The admitted sort: its plan, a private scratch directory, and the descriptor and concurrency permits.  |
| `SortSession`     | Push rows, `finish` into a `Stream` of values in key order. Spills and merges without caller involvement. |

A sort stays in memory only when its estimate fits under the configured
ceiling *and* the available memory can hold it. Otherwise the planner shares
the descriptor headroom across the sorts in flight (floored at two — a merge
needs two inputs), then sizes the run buffer so the run count stays within
`fan_in²` without exceeding available memory.
Under descriptor pressure it deliberately spends memory to keep pass depth
small. Incremental frontier compaction bounds metadata independently of that
estimate, with logarithmic merge depth instead of repeatedly rewriting the
whole accumulated input.

## Guarantees

- **Bounded memory.** A session buffers at most one run; the merge holds one
  head row per open reader. Spill metadata occupies 64 optional path slots,
  one per bit in the checked 64-bit run counter. Nothing materialises the
  whole input.
- **Bounded descriptors.** At most `max_fan_in` readers and one output writer are
  open per sort. The governor's semaphore rations reader permits across sorts.
- **Asynchronous throughout.** All spill and merge I/O is awaited over
  `async-fs-io`; no blocking calls run on Tokio worker threads.
- **Deterministic order.** Equal keys are emitted in run order; with dedup,
  the first row of each equal-key group survives.
- **Self-cleaning.** The scratch directory is created lazily on the first
  spill and removed when the output stream ends or is dropped — and also when
  a spilled session is dropped before `finish()` (an error-path bail-out
  cannot leak run files).
- **Fail fast.** I/O, encode, and decode failures surface as typed
  `SorterError` variants; there are no silent fallbacks. Run-counter exhaustion
  fails before creating another file. A failed or cancelled spill invalidates
  the session so later calls cannot produce partial results.

Row keys and values must implement `serde::Serialize` and
`serde::de::DeserializeOwned`; spilled runs are framed CBOR.

## License

Licensed under either of:

- Apache License, Version 2.0 ([`LICENSE-APACHE`](LICENSE-APACHE));
- MIT License ([`LICENSE-MIT`](LICENSE-MIT)).

## Links

[crates-badge]: https://img.shields.io/crates/v/sort-governor.svg
[crates-url]: https://crates.io/crates/sort-governor
[docs-badge]: https://docs.rs/sort-governor/badge.svg
[docs-url]: https://docs.rs/sort-governor
[ci-badge]: https://github.com/legra-ai/sort-governor/actions/workflows/ci.yml/badge.svg
[ci-url]: https://github.com/legra-ai/sort-governor/actions/workflows/ci.yml
[license-badge]: https://img.shields.io/crates/l/sort-governor.svg
[license-url]: https://github.com/legra-ai/sort-governor/blob/main/LICENSE-APACHE
[downloads-badge]: https://img.shields.io/crates/d/sort-governor.svg
[downloads-url]: https://crates.io/crates/sort-governor