FluxFlow.Components.Storage 3.0.0

FluxFlow.Components.Storage

Standalone storage nodes for FluxFlow: blockified put/get/query/delete over an injected IStorageStore. No engine required.

Each node is a self-contained TPL Dataflow processor built on FluxFlow.Nodes. Every message travels as a FlowMessage<T> envelope (payload + correlation id), so the correlation id flows request → result for free — and onto the Found/NotFound and Records branches. The host owns the IStorageStore lifetime and injects the opened store into each node; the nodes never open or dispose it (exactly like the HTTP node over an HttpClient).

Nodes

// The host owns the store; the node just borrows it.
IStorageStore store = ...;
await using var put = new StoragePutNode(store);

var request = FlowMessage.Create(new StoragePutRequest
{
    Collection = "items",
    Key = "a",
    Value = "one"
});
await put.Input.SendAsync(request);

var result = await put.Output.ReceiveAsync();      // FlowMessage<StorageResult>
// result.CorrelationId == request.CorrelationId   // the envelope carries it

Output, Found, NotFound, Records, Errors, and Events are all broadcast ports — link each to as many downstream consumers as you like.

Put

await using var put = new StoragePutNode(store, new StoragePutOptions
{
    Collection = "items",
    Mode = StorageWriteMode.Upsert,
    EmitStoredRecord = true,
    BoundedCapacity = 128
});

StoragePutNode consumes StoragePutRequest and emits StorageResult. Supported modes are Upsert, Create, and Replace. The request can override the node mode per item.

Get

await using var get = new StorageGetNode(store, new StorageGetOptions
{
    Collection = "items",
    IncludeExpired = false
});

StorageGetNode consumes StorageGetRequest and emits StorageResult on Output. Found records are also fanned to Found; missing records are also fanned to NotFound. A missing record is a normal result, not a processing error.

Query

await using var query = new StorageQueryNode(store, new StorageQueryOptions
{
    Collection = "items",
    Offset = 0,
    Limit = 100,
    IncludeExpired = false,
    EmitRecordsInResult = true,
    EmitRecordOutputs = true
});

StorageQueryNode consumes StorageQueryRequest and emits one StorageQueryResult on Output. The Records port emits each returned StorageRecord (as a FlowMessage<StorageRecord>) when EmitRecordOutputs is true. Requests can filter by collection, key prefix, exact-match attributes, stored time bounds, expired-record policy, offset, and limit.

Delete

await using var delete = new StorageDeleteNode(store, new StorageDeleteOptions
{
    Collection = "items",
    EmitMissingAsResult = true
});

StorageDeleteNode consumes StorageDeleteRequest and emits StorageResult. Missing deletes can be emitted as normal results or suppressed (EmitMissingAsResult).

Errors and events

A store failure or an invalid request surfaces a FlowError on Errors (stamped with the in-flight correlation id and a Code from StorageErrorCodes) and the node keeps processing later messages. Each node also emits FlowEvent diagnostics on Events (names in StorageDiagnosticNames).

Store Ownership

The package does not include a concrete database. A host supplies an IStorageStore — see the FileSystem and SqlFile adapter packages — and injects it into the nodes.

Adapter packages register a store factory through StorageComponentOptions:

var options = new StorageComponentOptions()
    .UseFileSystemStorage("./data");      // adapter extension
IStorageStore store = (await options.StoreFactory
    .OpenAsync(new StorageStoreContext { StoreName = "items-db" })).Store;

StorageStoreLease.Owned(store) marks a store the lease should dispose; StorageStoreLease.Shared(store) marks a host-owned store that must not be disposed. The factory receives the store name, default collection, and clock through StorageStoreContext.

Runtime Timing

Each node uses System.TimeProvider (default TimeProvider.System) for result timestamps. Pass a deterministic TimeProvider (for example Microsoft.Extensions.Time.Testing.FakeTimeProvider) when tests, replay, or deterministic dashboards need stable timestamps:

await using var put = new StoragePutNode(store, clock: fakeTimeProvider);

The same clock can be supplied to a backend store through StorageStoreContext.Clock so stored records and expiration checks share one time source.

Contracts

Core contracts:

• StoragePutRequest
• StorageGetRequest
• StorageQueryRequest
• StorageDeleteRequest
• StorageQueryResult
• StorageResult
• StorageRecord
• StorageWriteMode
• IStorageStore
• IStorageStoreFactory
• StorageStoreContext
• StorageStoreLease

StorageRecord.Value is object?: hosts own serialization and can compose this package with serialization or payload components before storage.

Composition Guidance

Use this package as one part of a host-composed graph. See Component Composition for recommended host boundaries, package boundaries, and extraction timing.