mostlylucid.ephemeral 1.3.0

Mostlylucid.Ephemeral

Fire... and Don't Quite Forget.

Bounded, observable, self-cleaning async execution with signal-based coordination.

dotnet add package mostlylucid.ephemeral

Quick Start

using Mostlylucid.Ephemeral;

// One-shot parallel processing
await items.EphemeralForEachAsync(
    async (item, ct) => await ProcessAsync(item, ct),
    new EphemeralOptions { MaxConcurrency = 8 });

// Long-lived coordinator
await using var coordinator = new EphemeralWorkCoordinator<WorkItem>(
    async (item, ct) => await ProcessAsync(item, ct),
    new EphemeralOptions { MaxConcurrency = 8 });

await coordinator.EnqueueAsync(new WorkItem("data"));

// See what's happening
var running = coordinator.GetRunning();
var failed = coordinator.GetFailed();
var pending = coordinator.PendingCount;

// Graceful shutdown
coordinator.Complete();
await coordinator.DrainAsync();

All Options

new EphemeralOptions
{
    // ═══════════════════════════════════════════════════════════════
    // CONCURRENCY
    // ═══════════════════════════════════════════════════════════════

    // Max parallel operations overall
    // Default: Environment.ProcessorCount
    MaxConcurrency = 8,

    // Allow runtime concurrency adjustment via SetMaxConcurrency()
    // Default: false (fastest hot-path)
    EnableDynamicConcurrency = false,

    // Max parallel operations per key (keyed coordinators only)
    // Default: 1 (sequential per key)
    MaxConcurrencyPerKey = 1,

    // ═══════════════════════════════════════════════════════════════
    // MEMORY / WINDOW
    // ═══════════════════════════════════════════════════════════════

    // Max operations retained in memory (LRU eviction)
    // Default: 200
    MaxTrackedOperations = 200,

    // Max age for tracked operations before cleanup
    // Default: 5 minutes
    MaxOperationLifetime = TimeSpan.FromMinutes(5),

    // ═══════════════════════════════════════════════════════════════
    // FAIR SCHEDULING (keyed coordinators only)
    // ═══════════════════════════════════════════════════════════════

    // Prevent hot keys from starving cold keys
    // Default: false (FIFO ordering)
    EnableFairScheduling = false,

    // Pending count before a key is deprioritized
    // Default: 10
    FairSchedulingThreshold = 10,

    // ═══════════════════════════════════════════════════════════════
    // SIGNALS
    // ═══════════════════════════════════════════════════════════════

    // Shared signal sink across coordinators
    // Default: null (isolated)
    Signals = new SignalSink(),

    // Sync callback on signal raise (keep fast!)
    // Default: null
    OnSignal = evt => Console.WriteLine($"Signal: {evt.Signal}"),

    // Async callback on signal raise (background queue, non-blocking)
    // Default: null
    OnSignalAsync = async (evt, ct) => await LogToService(evt, ct),

    // Sync callback on signal retract
    // Default: null
    OnSignalRetracted = evt => Console.WriteLine($"Retracted: {evt.Signal}"),

    // Async callback on signal retract
    // Default: null
    OnSignalRetractedAsync = async (evt, ct) => await NotifyService(evt, ct),

    // Max concurrent async signal handlers
    // Default: 4
    MaxConcurrentSignalHandlers = 4,

    // Max queued signals before dropping oldest
    // Default: 1000
    MaxQueuedSignals = 1000,

    // Self-documenting: signals this coordinator may emit
    // Default: null (no enforcement)
    Emits = new[] { "started", "completed", "error" },

    // Self-documenting: signals this coordinator listens for
    // Default: null (no enforcement)
    Listens = new[] { "backpressure", "shutdown" },

    // Constraints for signal propagation (cycles, depth, terminals)
    // Default: null
    SignalConstraints = new SignalConstraints { MaxDepth = 10 },

    // ═══════════════════════════════════════════════════════════════
    // SIGNAL-BASED CONTROL FLOW
    // ═══════════════════════════════════════════════════════════════

    // Skip new items when these signals are present
    // Use for circuit-breaker patterns
    // Default: null
    CancelOnSignals = new HashSet<string> { "circuit-open", "shutdown" },

    // Delay new items when these signals are present
    // Use for backpressure patterns
    // Default: null
    DeferOnSignals = new HashSet<string> { "backpressure", "rate-limited" },

    // How often to recheck when deferring
    // Default: 100ms
    DeferCheckInterval = TimeSpan.FromMilliseconds(100),

    // Max defer attempts before running anyway
    // Default: 50 (5 seconds at 100ms)
    MaxDeferAttempts = 50,

    // ═══════════════════════════════════════════════════════════════
    // SAMPLING
    // ═══════════════════════════════════════════════════════════════

    // Callback after each operation with window snapshot
    // Runs on caller's thread - keep it cheap!
    // Default: null
    OnSample = snapshot => metrics.Record(snapshot.Count)
}

Coordinators

EphemeralWorkCoordinator<T>

Long-lived work queue with bounded concurrency.

await using var coordinator = new EphemeralWorkCoordinator<Request>(
    async (req, ct) => await HandleAsync(req, ct),
    new EphemeralOptions { MaxConcurrency = 8 });

// Enqueue work
await coordinator.EnqueueAsync(request);
var id = await coordinator.EnqueueWithIdAsync(request);  // Get operation ID

// Query state
var running = coordinator.GetRunning();
var completed = coordinator.GetCompleted();
var failed = coordinator.GetFailed();
var snapshot = coordinator.GetSnapshot();
var byId = coordinator.GetById(id);

// Stats
int pending = coordinator.PendingCount;
int active = coordinator.ActiveCount;
int totalCompleted = coordinator.TotalCompleted;
int totalFailed = coordinator.TotalFailed;

// Dynamic concurrency (requires EnableDynamicConcurrency = true)
coordinator.SetMaxConcurrency(16);
int current = coordinator.CurrentMaxConcurrency;

// Signals
bool hasError = coordinator.HasSignal("error");
int errorCount = coordinator.CountSignals("error");
var errors = coordinator.GetSignalsByPattern("error.*");
var recent = coordinator.GetSignalsSince(DateTimeOffset.UtcNow.AddMinutes(-1));

// Shutdown
coordinator.Complete();          // Stop accepting new work
await coordinator.DrainAsync();  // Wait for in-flight to finish

EphemeralKeyedWorkCoordinator<T, TKey>

Per-key sequential processing with optional fair scheduling.

await using var coordinator = new EphemeralKeyedWorkCoordinator<Order, string>(
    keySelector: order => order.CustomerId,
    body: async (order, ct) => await ProcessOrder(order, ct),
    new EphemeralOptions
    {
        MaxConcurrency = 16,          // Total parallel
        MaxConcurrencyPerKey = 1,     // Sequential per customer
        EnableFairScheduling = true   // Prevent hot customer starvation
    });

await coordinator.EnqueueAsync(order);

EphemeralResultCoordinator<TInput, TResult>

Capture results from async operations.

await using var coordinator = new EphemeralResultCoordinator<Request, Response>(
    async (req, ct) => await FetchAsync(req, ct),
    new EphemeralOptions { MaxConcurrency = 4 });

var id = await coordinator.EnqueueAsync(request);
var snapshot = await coordinator.WaitForResult(id);

if (snapshot.HasResult)
    Console.WriteLine(snapshot.Result);
else if (snapshot.Exception != null)
    Console.WriteLine($"Failed: {snapshot.Exception.Message}");

PriorityWorkCoordinator<T>

Multiple priority lanes.

var coordinator = new PriorityWorkCoordinator<WorkItem>(
    body: async (item, ct) => await ProcessAsync(item, ct),
    new PriorityWorkCoordinatorOptions<WorkItem>(
        Lanes: new[]
        {
            new PriorityLane("critical", MaxDepth: 100),
            new PriorityLane("high"),
            new PriorityLane("normal"),
            new PriorityLane("low")
        }
    ));

await coordinator.EnqueueAsync(item, "critical");

Extension Method

One-shot parallel processing of collections.

await items.EphemeralForEachAsync(
    async (item, ct) => await ProcessAsync(item, ct),
    new EphemeralOptions
    {
        MaxConcurrency = 8,
        OnSignal = evt => Console.WriteLine(evt.Signal)
    });

Signals

Emitting Signals

// From within operation body (via ISignalEmitter passed to body)
await coordinator.EnqueueAsync(request);  // Body receives emitter

// From shared sink
var sink = new SignalSink();
sink.Raise("backpressure.downstream");
sink.Raise(new SignalEvent("error.timeout", operationId, key, DateTimeOffset.UtcNow));

// Retract signals
sink.Retract("backpressure.downstream");

Querying Signals

// Exact match
bool hasError = coordinator.HasSignal("error");
int count = coordinator.CountSignals("error");

// Pattern matching (glob-style: * and ?)
var errors = coordinator.GetSignalsByPattern("error.*");
var httpErrors = coordinator.GetSignalsByPattern("http.error.*");

// Time-based
var recent = coordinator.GetSignalsSince(DateTimeOffset.UtcNow.AddSeconds(-30));

// From shared sink
var snapshot = sink.Sense();
var filtered = sink.Sense(s => s.Signal.StartsWith("error"));

Shared Signal Sink

var sink = new SignalSink(maxCapacity: 1000, maxAge: TimeSpan.FromMinutes(5));

var coordinator1 = new EphemeralWorkCoordinator<A>(body, new EphemeralOptions { Signals = sink });
var coordinator2 = new EphemeralWorkCoordinator<B>(body, new EphemeralOptions { Signals = sink });

// Both coordinators see signals raised by either
sink.Raise("system.maintenance");

Dependency Injection

// Register
services.AddEphemeralWorkCoordinator<WorkItem>(
    async (item, ct) => await ProcessAsync(item, ct),
    new EphemeralOptions { MaxConcurrency = 8 });

// Named coordinators
services.AddEphemeralWorkCoordinator<WorkItem>("priority",
    async (item, ct) => await ProcessPriorityAsync(item, ct),
    new EphemeralOptions { MaxConcurrency = 4 });

// Inject
public class MyService(IEphemeralCoordinatorFactory<WorkItem> factory)
{
    public async Task DoWork()
    {
        var coordinator = factory.CreateCoordinator();
        // or: factory.CreateCoordinator("priority")
        await coordinator.EnqueueAsync(new WorkItem());
    }
}

Related Packages

Target Frameworks

.NET 6.0, 7.0, 8.0, 9.0, 10.0

License

Unlicense (public domain)