SpillFlow 0.3.0-alpha.9

SpillFlow

SpillFlow is an experimental .NET 8 library for bounded-memory operations over IAsyncEnumerable<T>. It provides OrderBySpill, DistinctSpill, ExceptBySpill, GroupJoinSpill, GroupJoinStreamSpill, DistinctRecoverableSpill, ExceptByRecoverableSpill, GroupAggregateSpill, GroupAggregateRecoverableSpill, and GroupJoinStreamRecoverableSpill.

Alpha: APIs, generated codec layout, and recovery file formats may change before 1.0. Review the current limitations before production use.

Why SpillFlow

• Processes data larger than available RAM through bounded temporary storage.
• Supports asynchronous Distinct, OrderBy, ExceptBy, grouped aggregation, materialized joins, and disk-streamed joins.
• Uses no third-party runtime packages.
• Provides disk quotas, bounded merge fan-in, recursive partitioning, checksums, schema fingerprints, optional AES-GCM encryption, and protected session folders.
• Includes recoverable operators and transactional output identifiers for restartable long-running jobs.
• Includes a source generator for deterministic compact binary codecs.

In the checked-in Windows x64 comparison, SpillFlow used 18.6% to 43.0% less median peak working set than ExternalSort across the four shared operations. These are measured results for the documented workload, not a universal promise.

Install

dotnet add package SpillFlow --version 0.3.0-alpha.9

The package includes the codec source generator; no second analyzer package is required. SpillFlow targets .NET 8.

The implementation uses an in-memory fast path for small distinct workloads, stable hash partitioning for set and join operations, and sorted runs with a k-way merge for ordering. A SpillSession owns temporary storage and removes it when disposed.

await using var session = new SpillSession(new SpillOptions
{
    MemoryLimitBytes = 256L * 1024 * 1024,
    MaxProcessPrivateMemoryBytes = 1024L * 1024 * 1024
});

session.RegisterCodec(new TransactionSpillCodec());

await foreach (var row in transactions.DistinctSpill(session, x => x.Id))
{
    // Process each unique transaction.
}

For production record types, annotate the type and register its generated codec:

[GenerateSpillCodec]
public sealed class Transaction
{
    public Guid Id { get; set; }
    public long AmountInCents { get; set; }
    public string Currency { get; set; } = "";
}

session.RegisterCodec(new TransactionSpillCodec());

The generator supports public get/set properties of type int, long, bool, Guid, and string. It emits checked, deterministic little-endian encoding and rejects unsupported property types at compile time. You can instead implement ISpillCodec<T> manually. int, long, and Guid have built-in fixed-width encoding. Without a codec, records use the slower JSON fallback.

Every binary codec declares a stable SchemaId. SpillFlow stores a 128-bit fingerprint of that identifier in each temporary block and rejects mismatched blocks before decoding. A custom codec must change its SchemaId whenever its binary layout changes.

Production-style callers can require binary codecs and reject the JSON fallback:

await using var strictSession = new SpillSession(new SpillOptions
{
    MemoryLimitBytes = 256L * 1024 * 1024,
    AllowJsonFallback = false
});

Grouped aggregation retains one accumulator per key instead of materializing every member of every group:

var totals = transactions.GroupAggregateSpill(
    session,
    row => row.AccountId,
    () => 0m,
    (total, row) => total + row.Amount,
    (accountId, total) => new AccountTotal(accountId, total));

GroupAggregateSpill begins with an adaptive in-memory path. It creates no temporary files when the estimated accumulator state fits within 75 percent of the session budget. If the threshold is crossed, it writes the buffered prefix to stable hash partitions and continues in bounded-memory mode without reading the source a second time.

GroupJoinSpill returns each match group as an IReadOnlyList, so a single group must fit its memory lease. For a key with more matches than RAM, use GroupJoinStreamSpill. Its result selector consumes an asynchronous match stream before returning:

var counts = accounts.GroupJoinStreamSpill(
    transactions,
    session,
    account => account.Id,
    transaction => transaction.AccountId,
    async (account, matches, cancellationToken) =>
    {
        var count = 0L;
        await foreach (var transaction in matches.WithCancellation(cancellationToken))
            count++;
        return new AccountTransactionCount(account.Id, count);
    });

When an oversized partition stops shrinking, the streaming operator reads its matches directly from disk. Left records are processed in bounded batches, and one right-partition scan broadcasts matches to every consumer in that batch through one-item channels with backpressure. HotKeyJoinBatchSize controls the maximum batch size and defaults to 64. HotKeyJoinBatches and HotKeyPartitionScans expose the actual work in session statistics. This keeps memory bounded and reduces scans from one per left record to approximately one per batch. Grouped aggregation remains preferable when callers need only a count, sum, or other summary.

Sorting is stable for equal keys and enforces MergeFanIn through bounded multi-pass merging. Hash operators use the supplied equality comparer for both equality and partition routing. Before a partition is materialized, SpillFlow applies PartitionMemoryExpansionFactor and fails with SpillResourceLimitException when the estimated retained state cannot fit safely inside the session budget.

MaxProcessPrivateMemoryBytes is an optional process-wide safety tripwire. SpillFlow samples private memory at session creation and at record or lease boundaries, no more often than ProcessMemoryCheckInterval, and reports the largest observation in PeakObservedProcessPrivateMemoryBytes. Sampling cannot guarantee a hard ceiling between checks. Use a container, Windows Job Object, or service-level operating-system limit when strict process isolation is needed.

For long-running sorting jobs whose source can reopen at a record offset, OrderByRecoverableSpill checkpoints input runs, merge replacements, and output progress:

await foreach (var row in session.OrderByRecoverableSpill(
    (offset, cancellationToken) => ReadTransactionsFromOffset(offset, cancellationToken),
    operationId: "daily-transactions",
    keySelector: transaction => transaction.Timestamp,
    outputCheckpointInterval: 1_000,
    cancellationToken: cancellationToken))
{
    await WriteResult(row, cancellationToken);
}

After an interruption, reopen the preserved directory with SpillSession.Recover and call the same operation ID. Reading resumes at the last durable run boundary, and merging resumes from the last durable run graph. Output is at-least-once: after a crash, up to outputCheckpointInterval records may be replayed. Use ProcessTransactionallyAsync when the destination can save the business change and a unique delivery acknowledgement in one transaction:

await session.OrderByRecoverableSpill(
        (offset, token) => ReadTransactionsFromOffset(offset, token),
        operationId: "daily-transactions",
        keySelector: transaction => transaction.Timestamp,
        cancellationToken: cancellationToken)
    .ProcessTransactionallyAsync(
        operationId: "post-daily-transactions",
        deliveryKeySelector: transaction => transaction.Id.ToString(),
        sink: sqlTransactionSink,
        cancellationToken: cancellationToken);

ISpillTransactionalSink<T> has two operations. IsAcknowledgedAsync checks a durable acknowledgement table. CommitAndAcknowledgeAsync must write the output and insert its (OperationId, DeliveryKey) acknowledgement under a unique constraint in the same database transaction. A lost response can cause SpillFlow to ask again, but the committed acknowledgement makes the retry a no-op. The delivery key must be unique for one logical output and stable across restarts; a database primary key or ledger transaction ID is a good choice. This gives exactly-once effects at that transactional destination. It cannot atomically cover two independent databases or a nontransactional external API.

Recoverable distinct processing uses the same offset-aware source contract:

await foreach (var row in session.DistinctRecoverableSpill(
    (offset, cancellationToken) => ReadTransactionsFromOffset(offset, cancellationToken),
    operationId: "unique-transactions",
    keySelector: transaction => transaction.Id,
    outputCheckpointInterval: 1_000,
    cancellationToken: cancellationToken))
{
    await WriteUniqueResult(row, cancellationToken);
}

It checkpoints durable input chunks before partitioning. Raw chunks are deleted only after the complete recursively refined partition graph is checkpointed, so interrupted partition construction can restart without reopening the source. Output is at-least-once and may replay up to the configured interval within the current partition. Recovery must use equivalent key-selector and comparer logic.

ExceptByRecoverableSpill accepts separate offset-aware factories for the primary sequence and exclusion keys. Each source advances independently. Both sets of input chunks remain available until the complete paired-partition graph is durably saved, so interrupted partitioning restarts locally. Output has the same at-least-once checkpoint window as recoverable distinct processing.

Recoverable grouped aggregation is intended for long-running summaries such as account balances, transaction counts, and ledger totals:

await foreach (var total in session.GroupAggregateRecoverableSpill(
    (offset, cancellationToken) => ReadTransactionsFromOffset(offset, cancellationToken),
    operationId: "daily-account-totals",
    keySelector: transaction => transaction.AccountId,
    seedFactory: () => 0L,
    accumulate: (sum, transaction) => sum + transaction.AmountInCents,
    resultSelector: (accountId, sum) => new AccountTotal(accountId, sum),
    cancellationToken: cancellationToken))
{
    await WriteTotal(total, cancellationToken);
}

The active partition is aggregated again after output interruption. Therefore the key selector, seed factory, accumulator, and result selector must be deterministic and free of external side effects. The downstream sink must handle the documented at-least-once replay window.

GroupJoinStreamRecoverableSpill provides the same two independent source offsets as recoverable ExceptBy while preserving the streaming hot-key path. If a right partition cannot fit after recursive refinement, a bounded batch of left records shares each disk scan and receives matches through asynchronous streams. Acknowledged left records are skipped without rerunning the result selector. The selector must consume the match stream before returning and tolerate being called again for an uncheckpointed output after interruption.

Status

This is a correctness-first prototype, not yet a production banking library. It now includes recursive partitioning, session memory leases, bounded merge fan-in, pooled asynchronous block I/O, disk quotas, abandoned-session cleanup, checksummed and schema-fingerprinted blocks, optional AES-GCM encryption, source-generated codecs, private per-user session directories, explicit caller checkpoints, operator-managed recovery for external ordering, recoverable distinct, ExceptBy, grouped aggregation, and streaming joins, plus optional process private-memory monitoring. The materialized-list join remains available for groups that fit RAM; recoverable workloads should use the streaming join. Remaining work includes secure-deletion policy and statistically repeated benchmarks on multiple storage devices. Transactional output integration is available when the destination implements ISpillTransactionalSink<T>.

Local failure qualification

The failure harness uses child processes and generated integer records. It does not call a hosted service or require a paid dataset:

dotnet build tools/SpillFlow.FailureHarness -c Release --no-restore
dotnet run -c Release --no-build --project tools/SpillFlow.FailureHarness

It hard-kills recoverable ordering during reading, merging, and output, then verifies the exact combined sequence after restart. It also requires fail-closed behavior for a corrupted block, missing run, truncated checkpoint, malformed manifest, and disk-quota exhaustion. A timestamped JSON evidence report is written under artifacts/failure-harness.

Security

Temporary blocks can be protected with a caller-managed AES-GCM key by setting SpillOptions.EncryptionKey. SpillFlow never stores or recovers the key. Do not hard-code production keys. Session directories use a protected current-user-only ACL on Windows and owner-only 0700 permissions on Unix by default. Permission setup fails closed; ProtectTemporaryDirectory = false is an explicit compatibility escape hatch. See the security policy and current limitations.

Contributing

See the contribution guide. This project uses the MIT License.

dotnet test SpillFlow.sln -c Release
dotnet run --project benchmarks/SpillFlow.Benchmarks -c Release -- 1000000

See BENCHMARK_RESULTS.md for the isolated 100-million-record SpillFlow versus ExternalSort matrix under a 1 GiB cap, plus a supplemental billion-row result. See the external memory comparison for the repeatable 10-million-record RAM gate and its measured scope.

For a simple but complete explanation of the architecture, operator algorithms, resource strategy, ExternalSort comparison, benchmark evidence, and current limits, see the architecture guide.

Documentation

• Architecture and algorithms
• Current limitations
• Security policy
• Internal security audit
• Memory comparison
• Changelog