Nun, das war meine Besessenheit für die vergangene Woche. Siehe die vorherigen Teile und was dazu führte; 'Was, wenn ein LRU ein Ausführungskontext war.'. Jetzt ist es ein Satz von 30 Nuget-Paketen, die die meisten wichtigen gleichzeitigen Ausführungsmuster (in TINY 5-10 Zeilenpaketen) abdecken.

Die Quelle finden Sie hier: https://github.com/scottgal/mostlylucid.atoms/blob/main/mostlylucid.ephemeral/src/mostlylucid.ephemeral.complete

Vorstrafen

Lesen Sie die [vorheriger Teil zu Signalen ]Hier ist die Readme.md aus den meistlucid.ephemeral.complete pacakage, die sowohl den Kern meistlucid.ephemeral Paket und alle Muster, 'Atome' (Koordinatoren etc) Pakete in einer bequemen DLL enthält.

Kern

ODER verwenden Sie den Kern meistlucid.ephemeral ein TINY (buchstäblich 10 Klassen), das Ihnen alle rohen Funktionen gibt.

Attribute & DI

ODER wenn Sie volle Attribut basierte async Routing mit einfachen [EphemeralJob] und service.AddKoordinator' Stil Registrierung verwenden die meistlucid.ephemeral.attributes Paket.

Dies ist wahrscheinlich das Thema meines Blogs geht weiter...Sie wurden gewarnt .

Meistlilucid.Ephemeral.Complete

Alle von Mostlylucid.Ephemeral in einer einzigen DLL - Eingeschränkte async-Ausführung mit signalbasierter Koordination.

dotnet add package mostlylucid.ephemeral.complete

Dieses Paket kompiliert alle Kern-, Atom- und Mustercodes in einer Assembly. Für einzelne Pakete, siehe die Links in jeder Abschnitt unten.

Schnellstart

using Mostlylucid.Ephemeral;

// Long-lived work 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"));

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

Registrierung von Dienstleistungen

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddCoordinator<WorkItem>(
    async (item, ct) => await ProcessAsync(item, ct),
    new EphemeralOptions { MaxConcurrency = 8, MaxTrackedOperations = 128 });

builder.Services.AddEphemeralSignalJobRunner<LogWatcherJobs>();

var app = builder.Build();
app.MapPost("/", async ([FromServices] IEphemeralCoordinatorFactory<WorkItem> factory, WorkItem item) =>
{
    var coordinator = factory.CreateCoordinator();
    await coordinator.EnqueueAsync(item);
    return Results.Accepted();
});

await app.RunAsync();

Das Vertraute services.AddCoordinator<T>() Helfer und AddEphemeralSignalJobRunner<T>() Halten Sie die Service-Registrierung kurz, lassen Sie DI das Waschbecken / den Starter besitzen, und machen Sie die neue Verantwortung / Cache / Logging Geschichten einen einzigen Klick entfernt.

Zugehörige Arbeitsplätze

mostlylucid.ephemeral.complete Bündel mostlylucid.ephemeral.attributes, so dass Attribut-Pipelines Teil des Kerns sind Oberfläche. Behandeln Sie den Läufer als erstklassigen Signalverbraucher: dekorierte Methoden verbinden das gleiche Caching, Protokollierung, und Pinning-Storys, und jedes Attribut kann deklarieren Priority, Arbeitsplatz MaxConcurrency, Lane, Key Quellen, Signal Emissionen, Pin/Expire Overrides und Retries.

Tasten für das Schlüsselattribut:

Bestellung & Fahrspuren: Verwendung Priority, MaxConcurrency, und Lane Arbeit in deterministischer Ordnung halten, während heiße Pfade Bleib getrennt.
Schlüssel & Tagging: OperationKey, KeyFromSignal, KeyFromPayload, und [KeySource] Hilfe bei der Gruppenarbeit mit sinnvolle Schlüssel für Protokollierung, faire Terminplanung und Diagnose.
Pinnen & Retries: Pin, ExpireAfterMs, AwaitSignals, MaxRetries, und RetryDelayMs die Handler ausdehnen lassen Ihre Sichtbarkeit, Gate-Ausführung, bis Abhängigkeiten ankommen, und heilen mit Retries beim Aussenden von Ausfallsignalen.
Signalchoreografie: Emit EmitOnStart, EmitOnComplete, und EmitOnFailure Um nachgelagerte Stufen zu signalisieren, loggen Beobachter oder andere Koordinatoren ohne manuelle Verkabelung.

var sink = new SignalSink();
await using var runner = new EphemeralSignalJobRunner(sink, new[] { new LogWatcherJobs(sink) });

var loggerFactory = LoggerFactory.Create(builder =>
{
    builder.AddConsole();
    builder.AddProvider(new SignalLoggerProvider(new TypedSignalSink<SignalLogPayload>(sink)));
});

var logger = loggerFactory.CreateLogger("orders");
logger.LogError(new EventId(1001, "DbFailure"), "Order store failed");

// Later tasks or other services can also raise watcher-friendly signals directly:
sink.Raise("log.error.orders.dbfailure", key: "orders");

public sealed class LogWatcherJobs
{
    private readonly SignalSink _sink;

    public LogWatcherJobs(SignalSink sink) => _sink = sink;

    [EphemeralJob("log.error.*", Priority = 1, MaxConcurrency = 2, Lane = "hot:4", EmitOnComplete = new[] { "incident.created" })]
    public Task EscalateAsync(SignalEvent signal)
    {
        Console.WriteLine($"escalating {signal.Signal} for {signal.Key}");
        _sink.Raise("incident.created", key: signal.Key);
        return Task.CompletedTask;
    }

    [EphemeralJob("incident.created", EmitOnStart = new[] { "incident.monitor.start" })]
    public Task NotifyAsync(SignalEvent signal)
    {
        Console.WriteLine($"notified incident for {signal.Key}");
        return Task.CompletedTask;
    }
}

Dieser Läufer sitzt jetzt beim Start und reagiert, wann immer log.error.* oder jedes emittierte Signal trifft die Senke. Attribut Handler können auch Schlüssel von Signalen/Payloads ablesen, Pin-Arbeiten bis nachgeschaltete Acks durchführen, Vervollständigung/Failure-Signale aussenden und Einschub in die Fahrspuren für die Bestellung. Für DI-erste Setups verwenden services.AddEphemeralSignalJobRunner<T>() (oder der Anwendungsbereich) Variante) so dass der Läufer und Senke werden durch den Container verwaltet.

[EphemeralJobs(SignalPrefix = "stage", DefaultLane = "pipeline")] öffentlich versiegelte Klasse StageJobs { [EphemeralJob("geringest", EmitOnComplete = neu[] { "stage.ingest.done" }] public Task IngestAsync(SignalEvent evt) => Console.Out.WriteLineAsync(evt.Signal);

[EphemeralJob("finalize")]
public Task FinalizeAsync(SignalEvent evt) => Console.Out.WriteLineAsync("final stage");

}

var stageSink = neuer SignalSink(); warten mit var stageRunner = neue EphemeralSignalJobRunner(stageSink, neue[] { neue StageJobs() }); stageSink.Raise("stage.ingest");

Pin-schwere Jobs können sich auf ResponsibilitySignalManager.PinUntilQueried (Standard-Ack-Muster) responsibility.ack.*) bis halten ihre Operationen sichtbar, bis ein nachgeschalteter Leser die Nutzlast abholt, während OperationEchoMaker/ OperationEchoAtom den letzten Signalstrom fortbestehen, so dass Auditoren oder Moleküle noch den letzten Zustand nach Das Atom stirbt.

Geplante Aufgaben

mostlylucid.ephemeral.complete enthält auch mostlylucid.ephemeral.atoms.scheduledtasks. Definieren Sie Cron oder JSON Zeitpläne über ScheduledTaskDefinition (Krone, Signal, optional) key, payload, description, timeZone, format, runOnStartup, usw.), und lassen Sie ScheduledTasksAtom enqueue dauerhafte Arbeit durch DurableTaskAtom. Jede geplante Aufgabe erhöht das konfigurierte Signal innerhalb eines Koordinatorfensters, so dass es Pinning, Protokollierung und Verantwortung Semantik erbt während Ihre Moleküle oder Attributpipelines auf die emittierte Signalwelle reagieren.

Jede DurableTask trägt den Zeitplan Name, Signal, fakultativ Key, sogar eine getippte Payload, und Description, so dass nachgeschaltete Hörer sofort wissen, welcher Job lief und welche Metadaten (Dateinamen, URLs, etc.) zu verbrauchen. DurableTaskAtom.WaitForIdleAsync() wenn Sie nur auf den aktuellen Burst der geplanten Arbeit warten wollen, ohne das Atom zu beenden, halten Sie den Scheduler bereit für die nächste Cron tick.

Protokollierung & Signale

mostlylucid.ephemeral.logging Spiegel Microsoft.Extensions.Logging in Signale und umgekehrt. Beginnen Sie mit der Befestigung SignalLoggerProvider zu Ihrer Logger-Fabrik so Log-Events erhöhen log.* Signale und Haken SignalToLoggerAdapter wenn Sie möchten, dass Signale in die Standard-Log-Pipeline zurückfließen.

var sink = new SignalSink();
var typedSink = new TypedSignalSink<SignalLogPayload>(sink);

using var loggerFactory = LoggerFactory.Create(builder =>
{
    builder.AddConsole();
    builder.AddProvider(new SignalLoggerProvider(typedSink));
});

using var watcher = new EphemeralSignalJobRunner(sink, new[] { new LogWatcherJobs(sink) });

var logger = loggerFactory.CreateLogger("orders");
logger.LogError(new EventId(1001, "DbFailure"), "Order store failed");

public sealed class LogWatcherJobs
{
    private readonly SignalSink _sink;

    public LogWatcherJobs(SignalSink sink) => _sink = sink;

    [EphemeralJob("log.error.*")]
    public Task EscalateAsync(SignalEvent signal)
    {
        _sink.Raise("incident.created", key: signal.Key);
        return Task.CompletedTask;
    }

    [EphemeralJob("incident.created")]
    public Task NotifyAsync(SignalEvent signal)
    {
        Console.WriteLine($"Incident for {signal.Key}");
        return Task.CompletedTask;
    }
}

Verwendung SignalToLoggerAdapter um die daraus resultierenden Signale wieder in Standard-Logs zu spiegeln, so dass Ihr Monitoring-Stack beides sieht Seite der Brücke.

Kernkoordinatoren

Packung: meistlucid.ephemeral

Ephemerer Arbeitskoordinator<T>

Langlebige Arbeitswarteschlange mit begrenzter Übereinstimmung und beobachtbarem Fenster.

await using var coordinator = new EphemeralWorkCoordinator<Request>(
    async (req, ct) => await HandleAsync(req, ct),
    new EphemeralOptions
    {
        MaxConcurrency = 8,
        MaxTrackedOperations = 200,
        MaxOperationLifetime = TimeSpan.FromMinutes(5)
    });

await coordinator.EnqueueAsync(request);

// Observe state
var running = coordinator.GetRunning();
var failed = coordinator.GetFailed();
var pending = coordinator.PendingCount;

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

EphemeralKeyedWorkKoordinator<TKey, T>

Pro-key sequentielle Verarbeitung - Elemente mit demselben Schlüssel in der Reihenfolge verarbeitet.

await using var coordinator = new EphemeralKeyedWorkCoordinator<Order, string>(
    order => order.CustomerId,  // Key selector
    async (order, ct) => await ProcessOrder(order, ct),
    new EphemeralOptions
    {
        MaxConcurrency = 16,      // Total parallel
        MaxConcurrencyPerKey = 1  // Sequential per customer
    });

await coordinator.EnqueueAsync(order);

EphemeralResultKoordinator<TInput, TResult>

Erfassen von Ergebnissen aus async-Operationen.

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);

Vorrangiger Arbeitskoordinator<T>

Mehrere Prioritätsspuren mit konfigurierbarer Koncurrenz pro Spur.

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

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

Konfiguration (EphemeraleOptionen)

new EphemeralOptions
{
    // Concurrency
    MaxConcurrency = 8,                    // Max parallel operations
    MaxConcurrencyPerKey = 1,              // For keyed coordinators
    EnableDynamicConcurrency = false,      // Allow runtime adjustment

    // Memory
    MaxTrackedOperations = 200,            // Window size (LRU eviction)
    MaxOperationLifetime = TimeSpan.FromMinutes(5),

    // Fair scheduling (keyed only)
    EnableFairScheduling = false,          // Prevent hot key starvation
    FairSchedulingThreshold = 10,

    // Signals
    Signals = sharedSink,                  // Shared signal sink
    OnSignal = evt => { },                 // Sync callback
    OnSignalAsync = async (evt, ct) => { }, // Async callback
    CancelOnSignals = new HashSet<string> { "circuit-open" },
    DeferOnSignals = new HashSet<string> { "backpressure" },
    DeferCheckInterval = TimeSpan.FromMilliseconds(100),
    MaxDeferAttempts = 50,

    // Signal handler limits
    MaxConcurrentSignalHandlers = 4,
    MaxQueuedSignals = 1000
}

Signale

Der Betrieb sendet Signale für die bereichsübergreifende Beobachtbarkeit aus.

// Query signals
bool hasError = coordinator.HasSignal("error");
int count = coordinator.CountSignals("error");
var errors = coordinator.GetSignalsByPattern("error.*");

// Shared sink across coordinators
var sink = new SignalSink();
var c1 = new EphemeralWorkCoordinator<A>(body, new EphemeralOptions { Signals = sink });
var c2 = new EphemeralWorkCoordinator<B>(body, new EphemeralOptions { Signals = sink });
sink.Raise("system.busy");  // Both see it

Verantwortung Signale & Finalisierung

Müssen die Ergebnisse für nachgelagerte Verbraucher nur lange genug sichtbar bleiben? ResponsibilitySignalManager können Sie pin eine Betrieb bis zum Eintreffen eines ack-Signals (Standardmuster) responsibility.ack.* mit Schlüssel=operationId). Bereitstellung einer fakultativ description so kann die Operation ihre Verantwortung beschreiben und festlegen maxPinDuration zu anmutig Selbst-klar, wenn der Verbraucher nie auftaucht.

var manager = new ResponsibilitySignalManager(coordinator, sink, maxPinDuration: TimeSpan.FromMinutes(5));
if (manager.PinUntilQueried(operationId, "file.ready", ackKey: fileId, description: "Awaiting fetch"))
{
    sink.Raise("file.ready", key: fileId);
}
// Consumer acknowledges the work
sink.Raise("file.ready.ack", key: fileId);

using Mostlylucid.Ephemeral.Patterns;

var notes = new LastWordsNoteAtom(async note => await noteRepository.SaveAsync(note));
coordinator.OperationFinalized += snapshot =>
{
    var note = new LastWordsNote(
        OperationId: snapshot.OperationId,
        Key: snapshot.Key,
        Signal: snapshot.Signals?.FirstOrDefault(),
        Timestamp: DateTimeOffset.UtcNow);

    _ = notes.EnqueueAsync(note);
};

LastWordsNote bleibt winzig (Operation id, Schlüssel, Signal, Zeitstempel), so dass Sie jeden minimalen Zustand aufnehmen können, den Sie interessieren bevor die Operation gesammelt wird.

Der Koordinator hält auch ein kurzlebiges Echo der Endsignale (aktiviert über EnableOperationEcho), dass Sie können inspizieren mit GetEchoes() wenn Sie die getrimmte Signalwelle erneut abspielen müssen, ohne den vollen Betrieb zu halten.

var recentErrors = coordinator.GetEchoes(pattern: "error.*")
    .Where(e => e.Timestamp > DateTimeOffset.UtcNow - TimeSpan.FromMinutes(1))
    .ToList();

if (recentErrors.Any())
    logger.LogWarning("Trimmed errors: {Count}", recentErrors.Count);

OperationEchoRetention und OperationEchoCapacity lassen Sie ausgleichen, wie viele Echos Sie halten und wie lange sie verweilen, so können Sie die letzten Wörter wieder abspielen, gerade lang genug, um die Oberflächendiagnostik.

Der Manager entpinnt automatisch, wenn das ack feuert, aber Sie können anrufen CompleteResponsibility(operationId) zur Beendigung der (z.B. auf Retries). OperationFinalized wenn das Fenster sie schneidet, so abonnieren Sie, wenn Sie ein endgültiges Signal, Log-Diagnostik, oder führen Sie die letzten Wörter.

Atome (Bausteine)

FixedWorkAtom

**Packung: ** meistlucid.ephemeral.atoms.fixedwork

Fixed Worker Pool mit Statistiken. Minimal API Wrapper um EphemeralWorkCoordinator.

using Mostlylucid.Ephemeral.Atoms.FixedWork;

await using var atom = new FixedWorkAtom<WorkItem>(
    async (item, ct) => await ProcessAsync(item, ct),
    maxConcurrency: 4,
    maxTracked: 200);

await atom.EnqueueAsync(item);

// Get stats
var (pending, active, completed, failed) = atom.Stats();
Console.WriteLine($"Completed: {completed}, Failed: {failed}");

// Get recent operations
var snapshot = atom.Snapshot();

// Graceful shutdown
await atom.DrainAsync();

GeschlüsseltSequentielles Atom

**Packung: ** meistlucid.ephemeral.atoms.keyedsequential

Per-key-Sequential-Bearbeitung mit optionaler fairer Terminplanung.

using Mostlylucid.Ephemeral.Atoms.KeyedSequential;

await using var atom = new KeyedSequentialAtom<Order, string>(
    keySelector: order => order.CustomerId,
    body: async (order, ct) => await ProcessOrder(order, ct),
    maxConcurrency: 16,
    perKeyConcurrency: 1,           // Sequential per key
    enableFairScheduling: true);    // Prevent hot key starvation

await atom.EnqueueAsync(order1);  // Customer A
await atom.EnqueueAsync(order2);  // Customer A - waits for order1
await atom.EnqueueAsync(order3);  // Customer B - parallel with A

var (pending, active, completed, failed) = atom.Stats();
await atom.DrainAsync();

SignalAwareAtom

**Packung: ** meistlucid.ephemeral.atoms.signalaware

Ansaugen oder Ansaugen aufgrund von Umgebungssignalen.

using Mostlylucid.Ephemeral.Atoms.SignalAware;

var sink = new SignalSink();

await using var atom = new SignalAwareAtom<WorkItem>(
    async (item, ct) => await ProcessAsync(item, ct),
    cancelOn: new HashSet<string> { "shutdown", "circuit-open" },
    deferOn: new HashSet<string> { "backpressure.*" },
    deferInterval: TimeSpan.FromMilliseconds(100),
    maxDeferAttempts: 50,
    signals: sink,
    maxConcurrency: 8);

// Enqueue work
await atom.EnqueueAsync(item);

// Raise ambient signals
atom.Raise("backpressure.downstream");  // New items defer
sink.Raise("shutdown");                  // New items rejected (returns -1)

await atom.DrainAsync();

BatchingAtom

**Packung: ** meistlucid.ephemeral.atoms.batching

Sammeln Sie Artikel in Chargen nach Größe oder Zeitintervall.

using Mostlylucid.Ephemeral.Atoms.Batching;

await using var atom = new BatchingAtom<LogEntry>(
    onBatch: async (batch, ct) =>
    {
        Console.WriteLine($"Flushing {batch.Count} entries");
        await FlushToDatabase(batch, ct);
    },
    maxBatchSize: 100,
    flushInterval: TimeSpan.FromSeconds(5));

// Items are batched automatically
atom.Enqueue(new LogEntry("User logged in"));
atom.Enqueue(new LogEntry("Request received"));
// ... batch flushes when full OR after 5 seconds

RetryAtom

Packung: meistlucid.ephemeral.atoms.retry

Exponentielle Backoff-Retry-Wrapper.

using Mostlylucid.Ephemeral.Atoms.Retry;

await using var atom = new RetryAtom<ApiRequest>(
    async (req, ct) => await CallExternalApi(req, ct),
    maxAttempts: 3,
    backoff: attempt => TimeSpan.FromMilliseconds(100 * Math.Pow(2, attempt)),
    maxConcurrency: 4);

// Automatically retries on failure with exponential backoff
// Attempt 1: immediate
// Attempt 2: 200ms delay
// Attempt 3: 400ms delay
await atom.EnqueueAsync(new ApiRequest("https://api.example.com"));

await atom.DrainAsync();

Datenspeicher-Atome

Packung: überwiegendlucid.ephemeral.atoms.data

Gemeinsame Konfiguration für Speicheratome (DataStorageConfig, IDataStorageAtom<TKey, TValue>) plus die Signalkonventionen, die Datei, SQLite und PostgreSQL-Adapter antreiben.

using Mostlylucid.Ephemeral.Atoms.Data;
using Mostlylucid.Ephemeral.Atoms.Data.File;

var sink = new SignalSink();
var config = new DataStorageConfig
{
    DatabaseName = "orders",
    SignalPrefix = "save.data",
    LoadSignalPrefix = "load.data",
    DeleteSignalPrefix = "delete.data",
    MaxConcurrency = 1
};

await using var storage = new FileDataStorageAtom<string, Order>(sink, config, "./orders");

storage.EnqueueSave("order-123", new Order { Id = "order-123", Total = 42.00m });
var loaded = await storage.LoadAsync("order-123");

Verwende das gleiche DataStorageConfig mit Mostlylucid.Ephemeral.Atoms.Data.Sqlite oder Mostlylucid.Ephemeral.Atoms.Data.Postgres Implementierungen für langlebige, signalgesteuerte Persistenz mit SQLite/Postgres. Attributaufträge können abonniert werden saved.data.{dbname} Signale, um nachgelagerte Arbeiten zu starten, während load.data.{dbname} löst Hydrat-Caches aus.

MoleculeRunner & AtomTrigger

**Packung: ** meistlucid.ephemeral.atoms.moleküle

Aufdrucke, die mit MoleculeBlueprintBuilder lassen Sie die Atome (Zahlung, Inventar, Versand, Benachrichtigung), die ausgeführt werden sollte, wenn ein Signal wie order.placed kommt. MoleculeRunner hört auf den Auslöser pattern, erzeugt ein gemeinsames MoleculeContext, und führt jeden Schritt, während Sie zum Starten / Vervollständigung Ereignisse abonnieren. AtomTrigger wenn das Signal eines Atoms ein anderes Koordinator oder Molekül starten soll.

var sink = new SignalSink();
var blueprint = new MoleculeBlueprintBuilder("order", "order.placed")
    .AddAtom(async (ctx, ct) => await paymentCoordinator.EnqueueAsync(ctx.TriggerSignal.Key!, ct))
    .AddAtom(async (ctx, ct) =>
    {
        ctx.Raise("order.payment.complete", ctx.TriggerSignal.Key);
        await inventoryCoordinator.EnqueueAsync(ctx.TriggerSignal.Key!, ct);
    })
    .Build();

await using var runner = new MoleculeRunner(sink, new[] { blueprint }, serviceProvider);
using var trigger = new AtomTrigger(sink, "order.payment.complete", async (signal, ct) =>
{
    await notificationCoordinator.EnqueueAsync(signal.Key!, ct);
});

sink.Raise("order.placed", key: "order-42");

Molecule-Schritte können zusätzliche Signale erhöhen (ctx.Raise("order.shipping.start")), so dass der Rest des Systems nimmt die Schlagstock.

SlidingCacheAtom

**Packung: ** meistlucid.ephemeral.atoms.slidingcache

Cache mit gleitendem Ablauf - Zugriff auf ein Ergebnis setzt seine TTL zurück.

using Mostlylucid.Ephemeral.Atoms.SlidingCache;

await using var cache = new SlidingCacheAtom<string, UserProfile>(
    async (userId, ct) => await LoadUserProfileAsync(userId, ct),
    slidingExpiration: TimeSpan.FromMinutes(5),
    absoluteExpiration: TimeSpan.FromHours(1),
    maxSize: 1000);

// First call: computes and caches
var profile = await cache.GetOrComputeAsync("user-123");

// Second call within 5 minutes: returns cached, resets TTL
var cached = await cache.GetOrComputeAsync("user-123");

// Try get without computation (still resets TTL on hit)
if (cache.TryGet("user-123", out var profile))
    Console.WriteLine(profile.Name);

// Get stats
var stats = cache.GetStats();
Console.WriteLine($"Entries: {stats.TotalEntries}, Hot: {stats.HotEntries}");

EphemeralLruCache

Packung: Kern (mostlylucid.ephemeral) — selbstoptimierender Cache mit gleitendem TTL bei jedem Treffer und erweitertem TTL für Heißschlüssel.

using Mostlylucid.Ephemeral;

var cache = new EphemeralLruCache<string, Widget>(new EphemeralLruCacheOptions
{
    DefaultTtl = TimeSpan.FromMinutes(5),
    HotKeyExtension = TimeSpan.FromMinutes(30),
    HotAccessThreshold = 3,
    MaxSize = 10_000,
    SampleRate = 5 // emit 1 in 5 signals
});

var widget = await cache.GetOrAddAsync("widget:42", async key =>
{
    var data = await LoadWidgetAsync(key);
    return data!;
});

// Stats and signals to see how the cache self-focuses on hot keys
var stats = cache.GetStats();              // hot/expired counts, size
var signals = cache.GetSignals("cache.*"); // cache.hot/evict/miss/hit

Tipp: MemoryCache kann für den gleitenden Ablauf konfiguriert werden, sendet aber niemals die heißen/kalten Signale aus oder erweitert TTL für Hotkeys. EphemeralLruCache ist die selbstoptimierende Voreinstellung im Core-Paket (und in SqliteSingleWriter) wann immer Sie möchten, dass sich der Cache auf den aktiven Arbeitssatz konzentriert.

Echo-Erzeuger

Packung: meistlucid.ephemeral.atoms.echo

Erfassen Sie die getippten letzten Wörter, die eine Operation emittiert, bevor sie getrimmt wird. Das Atom hält ein begrenztes Fenster des Signals Nutzlasten (entsprechend ActivationSignalPattern / CaptureSignalPattern) und wenn OperationFinalized Brände, die es erzeugt OperationEchoEntry<TPayload> Aufzeichnungen, die Sie per OperationEchoAtom<TPayload>.

var sink = new SignalSink();
var typedSink = new TypedSignalSink<EchoPayload>(sink);
var echoAtom = new OperationEchoAtom<EchoPayload>(async echo => await repository.AppendAsync(echo));

await using var coordinator = new EphemeralWorkCoordinator<JobItem>(ProcessAsync);
using var maker = coordinator.EnableOperationEchoing(
    typedSink,
    echoAtom,
    new OperationEchoMakerOptions<EchoPayload>
    {
        ActivationSignalPattern = "echo.capture",
        CaptureSignalPattern = "echo.*",
        MaxTrackedOperations = 128
    });

typedSink.Raise("echo.capture", new EchoPayload("order-1", "archived"), key: "order-1");

Attribute Jobs erhöhen einfach das eingegebene Signal mit jedem Zustand, den sie für kritisch halten, und der Hersteller hält den Arbeitssatz gebunden, während du das Echo beharrst.

Muster (Ready-to-Use)

SignalbasiertCircuitBreaker

**Packung: ** meistlucid.ephemeral.patterns.circuitbreaker

Stateless-Leistungsschalter mit Signalverlaufsfenster.

using Mostlylucid.Ephemeral.Patterns.CircuitBreaker;

var breaker = new SignalBasedCircuitBreaker(
    failureSignal: "api.failure",
    threshold: 5,
    windowSize: TimeSpan.FromSeconds(30));

// Check before making calls
if (breaker.IsOpen(coordinator))
{
    var retryAfter = breaker.GetTimeUntilClose(coordinator);
    throw new CircuitOpenException("Too many failures", retryAfter);
}

// Pattern matching variant
if (breaker.IsOpenMatching(coordinator, "error.*"))
    throw new CircuitOpenException("Error pattern detected");

// Get current failure count
int failures = breaker.GetFailureCount(coordinator);

Signalgesteuerter Rückdruck

**Packung: ** meistlucid.ephemeral.Muster.Rückdruck

Queue-Tiefe-Management mit automatischer Dependance auf Backpressure-Signalen.

using Mostlylucid.Ephemeral.Patterns.Backpressure;

var sink = new SignalSink();

await using var coordinator = SignalDrivenBackpressure.Create<WorkItem>(
    async (item, ct) => await ProcessAsync(item, ct),
    sink,
    maxConcurrency: 4);

// Enqueue work
await coordinator.EnqueueAsync(item);

// When downstream is slow
sink.Raise("backpressure.downstream");  // New work auto-defers

// When recovered
sink.Retract("backpressure.downstream"); // Work resumes

ControlledFanOut

**Packung: ** meistlucid.ephemeral.patterns.controlledfanout

Global + Pro-Key-Gating für kontrollierten Parallelismus.

using Mostlylucid.Ephemeral.Patterns.ControlledFanOut;

await using var fanout = new ControlledFanOut<string, Request>(
    keySelector: req => req.TenantId,
    body: async (req, ct) => await ProcessAsync(req, ct),
    maxGlobalConcurrency: 100,  // Total parallel across all tenants
    perKeyConcurrency: 5);      // Max 5 parallel per tenant

// Items for same tenant processed with limit
await fanout.EnqueueAsync(requestA);  // Tenant1
await fanout.EnqueueAsync(requestB);  // Tenant1 - waits if 5 already running
await fanout.EnqueueAsync(requestC);  // Tenant2 - parallel with Tenant1

await fanout.DrainAsync();

AdaptiveRateService

**Packung: ** meistlucid.ephemeral.patterns.adaptiverate

Signalgesteuerte Geschwindigkeitsbegrenzung mit automatischem Rückwärtsgang.

using Mostlylucid.Ephemeral.Patterns.AdaptiveRate;

await using var service = new AdaptiveRateService<ApiRequest>(
    async (req, ct) => await CallApiAsync(req, ct),
    maxConcurrency: 8);

// Process with automatic rate limit handling
await service.ProcessAsync(request);

// When API returns 429, emit signal with retry-after
// Signal: "rate-limit:500ms"
// Service auto-parses and delays

Console.WriteLine($"Pending: {service.PendingCount}, Active: {service.ActiveCount}");

DynamischeKonkurrenzDemo

**Packung: ** meistlucid.ephemeral.patterns.dynamicconcurrency

Laufzeitkoncurrenz-Skalierung basierend auf Lastsignalen.

using Mostlylucid.Ephemeral.Patterns.DynamicConcurrency;

var sink = new SignalSink();

await using var demo = new DynamicConcurrencyDemo<WorkItem>(
    async (item, ct) => await ProcessAsync(item, ct),
    sink,
    minConcurrency: 2,
    maxConcurrency: 32,
    scaleUpPattern: "load.high",
    scaleDownPattern: "load.low");

await demo.EnqueueAsync(item);

// Concurrency adjusts automatically based on signals
sink.Raise("load.high");  // Concurrency doubles (up to max)
sink.Raise("load.low");   // Concurrency halves (down to min)

Console.WriteLine($"Current concurrency: {demo.CurrentMaxConcurrency}");

await demo.DrainAsync();

KeyedPriorityFanOut

**Packung: ** Hauptsächlichlucid.ephemeral.patterns.keyedpriorityfanout

Prioritätsspuren mit Pro-Key-Bestellung erhalten.

using Mostlylucid.Ephemeral.Patterns.KeyedPriorityFanOut;

await using var fanout = new KeyedPriorityFanOut<string, UserCommand>(
    keySelector: cmd => cmd.UserId,
    body: async (cmd, ct) => await HandleCommand(cmd, ct),
    maxConcurrency: 32,
    perKeyConcurrency: 1,  // Sequential per user
    maxPriorityDepth: 100);

// Normal lane
await fanout.EnqueueAsync(normalCommand);

// Priority lane - jumps the queue for that user
bool accepted = await fanout.EnqueuePriorityAsync(urgentCommand);

// Check lane depths
var counts = fanout.PendingCounts;
Console.WriteLine($"Priority: {counts.Priority}, Normal: {counts.Normal}");

await fanout.DrainAsync();

ReaktivFanOutPipeline

**Packung: ** meistlucid.ephemeral.patterns.reaktivfanout

Zweistufige Pipeline mit automatischem Gegendruck.

using Mostlylucid.Ephemeral.Patterns.ReactiveFanOut;

await using var pipeline = new ReactiveFanOutPipeline<WorkItem>(
    stage2Work: async (item, ct) => await SlowProcessing(item, ct),
    preStageWork: async (item, ct) => await FastPreprocessing(item, ct),
    stage1MaxConcurrency: 8,
    stage1MinConcurrency: 1,
    stage2MaxConcurrency: 4,
    backpressureThreshold: 32,  // Throttle when stage2 has 32+ pending
    reliefThreshold: 8);        // Resume when stage2 drops below 8

await pipeline.EnqueueAsync(item);

// Stage1 auto-throttles when stage2 backs up
Console.WriteLine($"Stage1 concurrency: {pipeline.Stage1CurrentMaxConcurrency}");
Console.WriteLine($"Stage2 pending: {pipeline.Stage2Pending}");

await pipeline.DrainAsync();

SignalAnomalyDetektor

**Packung: ** meistlucid.ephemeral.patterns.anomalydetektor

Bewegungs-Fenster-Anomalie-Erkennung.

using Mostlylucid.Ephemeral.Patterns.AnomalyDetector;

var sink = new SignalSink();

var detector = new SignalAnomalyDetector(
    sink,
    pattern: "error.*",
    threshold: 5,
    window: TimeSpan.FromSeconds(10));

// Check for anomalies
if (detector.IsAnomalous())
{
    Console.WriteLine("Anomaly detected! Too many errors.");
    TriggerAlert();
}

// Get current match count
int errorCount = detector.GetMatchCount();
Console.WriteLine($"Errors in window: {errorCount}");

SignalkoordinierteLesungen

**Packung: ** meistlucid.ephemeral.patterns.signalkoordinierteLesungen

Quiesce liest während Updates ohne harte Schlösser.

using Mostlylucid.Ephemeral.Patterns.SignalCoordinatedReads;

// Run demo: readers pause when update signal is present
var result = await SignalCoordinatedReads.RunAsync(
    readCount: 10,
    updateCount: 1);

Console.WriteLine($"Reads: {result.ReadsCompleted}, Updates: {result.UpdatesCompleted}");
Console.WriteLine($"Signals: {string.Join(", ", result.Signals)}");

// Manual implementation:
var sink = new SignalSink();

await using var readers = new EphemeralWorkCoordinator<Query>(
    body,
    new EphemeralOptions
    {
        DeferOnSignals = new HashSet<string> { "update.in-progress" },
        Signals = sink
    });

// Readers auto-defer when update is running
sink.Raise("update.in-progress");  // Readers wait
sink.Raise("update.done");         // Readers resume

SignalisierungHttpClient

**Packung: ** überwiegendlucid.ephemeral.patterns.signalinghttp

HTTP-Client mit Fortschrittssignalen.

using Mostlylucid.Ephemeral.Patterns.SignalingHttp;

var httpClient = new HttpClient();
var request = new HttpRequestMessage(HttpMethod.Get, "https://example.com/large-file");

// Create an emitter from your coordinator
// (emitter is any ISignalEmitter - operations implement this)

byte[] data = await SignalingHttpClient.DownloadWithSignalsAsync(
    httpClient,
    request,
    emitter);

// Signals emitted during download:
// - stage.starting
// - progress:0
// - stage.request
// - stage.headers
// - stage.reading
// - progress:25, progress:50, progress:75, progress:100
// - stage.completed

SignalLogWatcher

**Packung: ** meistlucid.ephemeral.patterns.signallogwatcher

Beobachten Sie das Signalfenster für Muster und lösen Sie Rückrufe aus.

using Mostlylucid.Ephemeral.Patterns.SignalLogWatcher;

var sink = new SignalSink();

await using var watcher = new SignalLogWatcher(
    sink,
    onMatch: evt =>
    {
        Console.WriteLine($"Error detected: {evt.Signal} at {evt.Timestamp}");
        AlertOps(evt);
    },
    pattern: "error.*",
    pollInterval: TimeSpan.FromMilliseconds(200));

// Watcher runs in background, calling onMatch for each new error signal
sink.Raise("error.database");    // -> onMatch called
sink.Raise("error.timeout");     // -> onMatch called
sink.Raise("info.started");      // -> ignored (doesn't match pattern)

TelemetrieSignalHandler

**Packung: ** meistlucid.ephemeral.patterns.telemetrie

OpenTelemetry/Application Insights Integration.

using Mostlylucid.Ephemeral.Patterns.Telemetry;

// Use in-memory for testing, or implement ITelemetryClient for real telemetry
var telemetry = new InMemoryTelemetryClient();

await using var handler = new TelemetrySignalHandler(telemetry);

// Wire up to coordinator
var options = new EphemeralOptions
{
    OnSignal = signal => handler.OnSignal(signal)
};

// Signals are processed asynchronously
// - "error.*" signals -> TrackExceptionAsync
// - "perf.*" signals -> TrackMetricAsync
// - all signals -> TrackEventAsync

Console.WriteLine($"Queued: {handler.QueuedCount}");
Console.WriteLine($"Processed: {handler.ProcessedCount}");
Console.WriteLine($"Dropped: {handler.DroppedCount}");

// Check recorded events
var events = telemetry.GetEvents();

LongWindowDemo

**Packung: ** meistlucid.ephemeral.patterns.longwindowdemo

Zeigt große Fensterkonfiguration für Audit-Trails.

using Mostlylucid.Ephemeral.Patterns.LongWindowDemo;

// Configure coordinator with large tracking window
var options = new EphemeralOptions
{
    MaxTrackedOperations = 10000,
    MaxOperationLifetime = TimeSpan.FromHours(24)
};

SignalReactionShowcase

**Packung: ** meistlucid.ephemeral.patterns.signalreaktionshowcase

Zeigt Signal-Sendungsmuster und Rückrufe.

using Mostlylucid.Ephemeral.Patterns.SignalReactionShowcase;

// See source for signal dispatch examples
// Demonstrates OnSignal, OnSignalAsync, CancelOnSignals, DeferOnSignals

PersistentSignalWindow

**Packung: ** meistlucid.ephemeral.patterns.persistentFenster

Signalfenster mit SQLite Persistenz - überlebt den Neustart des Prozesses.

using Mostlylucid.Ephemeral.Patterns.PersistentWindow;

await using var window = new PersistentSignalWindow(
    "Data Source=signals.db",
    flushInterval: TimeSpan.FromSeconds(30));

// On startup: restore previous signals
await window.LoadFromDiskAsync(maxAge: TimeSpan.FromHours(24));

// Raise signals as normal
window.Raise("order.completed", key: "order-service");
window.Raise("payment.processed", key: "payment-service");

// Query signals
var recentOrders = window.Sense("order.*");

// Signals automatically flush every 30 seconds
// Also flushes on dispose

// Get stats
var stats = window.GetStats();
Console.WriteLine($"In memory: {stats.InMemoryCount}, Flushed: {stats.LastFlushedId}");

Abhängigkeitsinjektion

// Register in Startup/Program.cs
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));

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

Moderne DI-Wurzeln bevorzugen möglicherweise die kürzeren Helfer wie services.AddCoordinator<T>(...), services.AddScopedCoordinator<T>(...), oder services.AddKeyedCoordinator<T, TKey>(...) da sie wie normal lesen AddX Registrierungen; sie delegieren einfach zu den Ephemer-spezifischen Helfern unter der Haube.

Zielrahmen

.NET 6.0, 7.0, 8.0, 9.0, 10.0

Lizenz

Unlizenz (öffentlicher Bereich)

meistlucid.ephemeral.complete; ein seltsames gleichzeitiges Systemmustersystem in einem LRU-Cache.

Vorstrafen

Kern

Attribute & DI

Meistlilucid.Ephemeral.Complete

Inhaltsverzeichnis

Schnellstart

Registrierung von Dienstleistungen

Zugehörige Arbeitsplätze

Geplante Aufgaben

Protokollierung & Signale

Kernkoordinatoren

Ephemerer Arbeitskoordinator<T>

EphemeralKeyedWorkKoordinator<TKey, T>

EphemeralResultKoordinator<TInput, TResult>

Vorrangiger Arbeitskoordinator<T>

Konfiguration (EphemeraleOptionen)

Signale

Verantwortung Signale & Finalisierung

Atome (Bausteine)

FixedWorkAtom

GeschlüsseltSequentielles Atom

SignalAwareAtom

BatchingAtom

RetryAtom

Datenspeicher-Atome

MoleculeRunner & AtomTrigger

SlidingCacheAtom

EphemeralLruCache

Echo-Erzeuger

Muster (Ready-to-Use)

SignalbasiertCircuitBreaker

Signalgesteuerter Rückdruck

ControlledFanOut

AdaptiveRateService

DynamischeKonkurrenzDemo

KeyedPriorityFanOut

ReaktivFanOutPipeline

SignalAnomalyDetektor

SignalkoordinierteLesungen

SignalisierungHttpClient

SignalLogWatcher

TelemetrieSignalHandler

LongWindowDemo

SignalReactionShowcase

PersistentSignalWindow

Abhängigkeitsinjektion

Zielrahmen

Lizenz