Det här har varit min obsession under den senaste veckan. tidigare delar och vad som ledde till dettaTänk om en LRU var ett avtryckskontext. Nu är det ' en uppsättning | 30 | Nuget-packningar som täcker de flesta större samtida implementeringsmönsterna |( | i TINY | МSK4 | linjära packager ♫ ). | få fantastiska anpassningsförmågan med en enkel syntax ♫

Hitta källan här.

Priorer

Lees den den tidigare delen om Signaler för att få en insikt i det's användningsområden . Här visas Readme.md från den mestlucideradeM SK3ephemeralMSC4kompletna pacakage som består av både kärnans mestlucidaMska5efemerala paket och alla mönsterMske6 M Ska7atomsMka8 |Mska9koordinatorer och så vidare~Mska10 paket i ett bekvämt DLL

Korn

Eller använda kärnan mestlucid.ephemeral en TINY ( bokstavligt talat 10 klassar) som ger dig all den råfunktionaliteten

Attribute & DI

Eller om du vill ha ett komplett attribut baserat på async-riktning med enkel [EphemeralJob] och tjänst.AttaKoordinatorn` använder mestlucid.ephemeral.Attributes package.

Det här är troligtvis det ämne som min blogg kommer att handla om.

Mostlylucid.Ephemeral.Komplett

Meta-package som referenserar alla Mostlylucid.Ephemeral packages - begränsad async exécution med signal

dotnet add package mostlylucid.ephemeral.complete

Det här är ett metapackage som tar in alla corepackningar, atomer och mönster som beroende. individuella paketreferenser - NuGet kommer att lösa till en enda version utan Namespace Collisions

Vad är nytt med 2.0

Att bryta mot förändringar

SignalSink-abonnementsmodell ändrats:

// OLD (1.x) - Event-based
sink.SignalRaised += handler;

// NEW (2.0) - Lock-free Subscribe() returning IDisposable
using var sub = sink.Subscribe(handler);

Det kompletta paketet är nu en meta

Förr kompilerade alla källor till en enda DLL (riskiganamespace-kollisier)
Nu referenserar enskilda paket som beroende
Säker att blanda med individuella paketreferenser i samma lösning

Nya egenskaper

Ledger-detektorsystem (Mostlylucid.Ephemeral.Atoms.Taxonomy.Ledger)
- DetectionLedger - Samlar mätarens bidrag med sigmoid aggregation
- DetectionContribution - Fabrikmetoder Bot(), Human(), Info(), VerifiedBot(), VerifiedGoodBot()
- CategoryScore.TotalWeight egendom ( som är ombyggd från Weight)
Signalpropagationsspårning - Cykldetektor och djupgränsning för signalkedjor
källan - Debug in i bibliotekskällan direkt från din IDE

Se docs/Taxonomy.md för det delade vokabulära kring substratet och eskalation.

Vinnig Start

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

Registering av tjänster

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

Den välkända services.AddCoordinator<T>() hjälper och AddEphemeralSignalJobRunner<T>() behålla tjänstesregistrering sammanfattad, låta DI äga sinken / löpare , och göra det nya ansvaret M SK3 kassera MSC4 att lagra historier med en enda klix bort

Attribute-driven jobb

mostlylucid.ephemeral.complete paket mostlylucid.ephemeral.attributes, så attribut Pipelines är en del av kärnan ytan. Treta löparen som en första sorts signalkonsument pinning stories, och varje attribut kan Priority, jobb MaxConcurrency, Lane, Key källor, signal utsläppen , pin/dröjda överrider

Nymmer på nyckelattributen:

Ordning & vägar: Använd Priority, MaxConcurrency, och Lane att hålla arbetet i deterministisk ordning medan de varma vägerna stanna separat.
Att märka &: OperationKey, KeyFromSignal, KeyFromPayload, och [KeySource] hjälpa er att arbeta i grupp med meningsfulla nyckeln för att registrera
Pinning & återspeglar: Pin, ExpireAfterMs, AwaitSignals, MaxRetries, och RetryDelayMs låta handledare utöka dess synlighet, öppningsoperation tills beroenden kommer fram, och läker med tillbakadrag medan de utsätter felsignalerM SK2
Signalkorreografi: Emit EmitOnStart, EmitOnComplete, och EmitOnFailure att signalera nedströms stadier, log bekymmer, eller andra koordinatorer utan handrör

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

Den här löparen sitter nu vid start och reagerar närhelst log.error.* eller någon utsläppt signal träffar anläggningen användare kan också läsa nyckeln från signaler slot in in lanes for ordering. För DI -first setups use services.AddEphemeralSignalJobRunner<T>() ( eller den skalerade variant) så att springaren och dammen sköts av behållaren.

[Äphämerliga jobb(SignalPräfix = "stadiumM SK3 Standardbann ♫ ♫ = ♫ offentligt hyllde klass StageJobs { [EphemeralJob("ingestM SK1 EmitOnComplete = ny[] { offentliga uppgifter IngestAsync(SignalEvent evt) M SK2 KonsoleMST3OutM ST4WriteLineA sync MST5evtM st6SignaalMst7

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

}

var scenSink = ny SignalSinkM SK1 vänta med var stageRunner = ny EphemeralSignalJobRunnerM SK1stageSink, ny[] { nya StageJobs scenSink.

Pin-svåra jobb kan lita på ResponsibilitySignalManager.PinUntilQueried (Standardack-mönster responsibility.ack.*) till hålla sina operationer synliga tills en nedströmsreader tar upp payload, medan OperationEchoMaker/ OperationEchoAtom förbli kvar i den slutgiltiga signalströmmen så att auditörer eller molekyler fortfarande kan “taste atomen dör.

Avsedda uppgifter

mostlylucid.ephemeral.complete består också av mostlylucid.ephemeral.atoms.scheduledtasks. Definiera cron eller JSON planer via ScheduledTaskDefinition ( key, payload, description, timeZone, format, runOnStartup, etc .), och släpp ScheduledTasksAtom kräva ett hållbart arbete genom DurableTaskAtom. Varje avsedda jobb lyfter det konfigurerade signalet inuti ett koordinatorfenster, så att det härvt pinning, ,, loging och ansvarsemantik medan dina molekyler eller attribut Pipeliner reagerar på den utsläppta signalvågen.

Varje DurableTask bär på schemat Name, Signal, optional Key, även ett tryckt Payload, och Description, så nedströms lyssnare vet omedelbart vilket jobb löstes och vilka metadata de ska använda. DurableTaskAtom.WaitForIdleAsync() när du bara vill vänta tills den nuvarande sprängningen av av avsedda arbetet slutar utan att färdigställa atomen, för att hålla skedularen redo för nästa cron-tick

Logging & Signaler

mostlylucid.ephemeral.logging spegel Microsoft.Extensioner.Anslut till signaler och vice versaM SK2 Börja med att ansluta SignalLoggerProvider till din loggerfabrik så registrera händelser log.* signaler, och hamn SignalToLoggerAdapter om du vill att signaler flödar tillbaka i standardprotokollet

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

Använda SignalToLoggerAdapter att speglar ut de utfallende signalerna tillbaka i standardprotokoll så att din övervakningsstapel ser båda sidor av brot.

Koordinatorer

Paketet: mestlucid.ephemeral

EphemeralWorkCoordinator<T>

lång-liv arbetskväva med begränsad samtida och synligt fönster

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

EphemeralKeyedWorkCoordinator<Tkey, T>

Per-"key sequential processing - items with same key processed in order

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

Koordinator för tidiga resultat<TInput , resultat>

Ta upp resultat från async operationer.

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

Prioritetsarbetande koordinator<T>

Flera prioritära banor med konfigurerade samtida banor per bana.

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 (EphemeralOptions)

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
}

Signaler

Operationer släpper ut signaler för korsning-skjutningsobservativitetM SK1

// 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

ansvarssignaler & Finalisering

Man måste hålla resultaten synliga tillräckligt länge för nedströmmade konsumenter ResponsibilitySignalManager låter dig sätta på en operation tills en acksignal kommer (default-mönster responsibility.ack.* med nyckel=operationId). Ge ett optional description så att operationen kan beskriva dess ansvar maxPinDuration på ett elegant sätt själv-klar om konsumenten aldrig dyker upp

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 stannar litet (operationsid, nyckeln , signal, , tidstämp,), så att du kan spela in vilken minimala tillstånd som helst du bryr dig om ungefär innan operationen samlas in.

Koordinatorn håller också ett kort echo av de slutgiltiga signalerna ( möjliga via EnableOperationEcho) att du kan att undersök med GetEchoes() när du behöver spela om den trimmerade signalvågen utan att hålla hela operationen runt

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 och OperationEchoCapacity låter dig balansera hur många ekor du håller och hur länge de stannar så att du kan spela upp de sista orden, bara tillräckligt länge för att kunna diagnostisera på ytan.

Chefen avsperar automatiskt när akuten aktiveras, men du kan ringa CompleteResponsibility(operationId) för att avsluta ansvarsfullhet tidigt (e.g ., på återflygningarM SK3 Operationer stiger fortfarande OperationFinalized när fönstret trimmar dem, så signera om du vill utsätta ett slutgiltiga signal, logdiagnostiken, eller köra M SK2 sista ord ” städa upp

Atomer (Blöcke

FixedWorkAtom

**Paketet: ** mestlucid.ephemeral.atomsM SK2fixedwork

Fixerad arbetspool med statistik. Minimal API-wrapper runt 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();

KeyedSequentialAtom

**Paketet: ** mestlucid.ephemeral.atomsM SK2keyedsequential

Per-key sequential processing with optional fair scheduling

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

**Paketet: ** mestlucid.ephemeral.atomsM SK2signalaware

pausa eller avbryta ingången baserat på omgivningssignaler.

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

**Paketet: ** mestlucid.ephemeral.atomsM SK2batching

Samma saker i paster efter storlek eller tidsinterval.

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

Försök om atom

Paketet: mestlucid.ephemeral.atomsM SK2retry

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

Data lagring atomer

Paketet: merlucid data

Samlad konfiguration för lagring av atomer (DataStorageConfig, IDataStorageAtom<TKey, TValue>) plus signalen konventioner som driver filen, SQLite , och PostgreSQL-adaptörer

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

Använd samma DataStorageConfig med Mostlylucid.Ephemeral.Atoms.Data.Sqlite eller Mostlylucid.Ephemeral.Atoms.Data.Postgres implementeringar för hållbarhet saved.data.{dbname} signaler för att starta nedströmsarbete medan load.data.{dbname} triglar vattenreser.

MoleculeRunner & AtomTrigger

**Paketet: ** mest korrekta molekyler

Blueprints komponerade med MoleculeBlueprintBuilder så kan du definiera atomerna. notification) som borde fungera när en signal som order.placed kommer fram. MoleculeRunner lyssnar på aktiveringsknappen mönster, skapar en gemensam MoleculeContext, och löser varje steg medan du anmäler dig för att starta/kollektivitetshändelser . Använd AtomTrigger när en atoms signal ska starta en annan koordinator eller molekyl

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

Molekulära steg kan höja ytterligare signaler (ctx.Raise("order.shipping.start")) så resten av systemet plockar upp slagtränken.

SlidingCacheAtom

**Paketet: ** mestlucid.ephemeral.atomsM SK2slidingcache

Cache med slättrande utgångsdatum - som när man går in på ett resultat återställer dess TTL .

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

VolatileOperationAtom

**Paketet: ** mestlucid.ephemeral.atomsM SK2volatil

Släpp operationer så fort de släpper ut ett dödssignal så att fönsteret håller sig lätta även under väldigt hög dragkraft atom lyssnar på ett konfigurerat mönster (default kill.*) på den delade SignalSink, samtal IOperationEvictor.TryKill med den fångade operationen ID, och ändå låter koordinatorn skjuta upp sina echo hooks before the entry försvinner.

var sink = new SignalSink();
await using var coordinator = new EphemeralWorkCoordinator<JobItem>(
    async (job, ct) => await QuickProcessAsync(job, ct),
    new EphemeralOptions
    {
        Signals = sink,
        EnableOperationEcho = true,
        OperationEchoRetention = TimeSpan.FromSeconds(30)
    });

using var volatileAtom = new VolatileOperationAtom(sink, coordinator);

// inside the job: emitter.Emit("kill.job");

Koppla atomen med OperationEchoMaker/OperationEchoAtom så skrivna *.echo.* signaler överlever döden som en liten TTL’d rekord.

EphemeralLruCache

Paketet: kärn (mostlylucid.ephemeral) ♫ — ♫ själv ♫ ♫- ♫ optimera kaset med skiftande TTL på varje träffande och utsträckta TTL för hot keys.

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 kan konfigureras för att glida utfallet, men den aldrig släpper ut de varma/kylna signalerna eller utökar TTL för hotkeys. EphemeralLruCache är den självoptimerande standarden i kärnpaketet ( och i SqliteSingleWriter) när du vill att kasken fokuserar på den aktiva arbetsgruppen.

Atom vid fönstrets storlek

**Paketet: ** mestlucid.ephemeral.atomsM SK2windowsize

Atomen lyssnar efter window.size.* eller window.time.* kommandor, klampar upp den efterfrågade kapaciteten /behållandeM SK2 och ropar SignalSink.UpdateWindowSize så att du kan utöka eller minska det delade vensteret från en annan koordinator eller signalvakter.

var sink = new SignalSink(maxCapacity: 250);
await using var atom = new WindowSizeAtom(sink);

sink.Raise("window.size.decrease:50");
sink.Raise("window.time.set:00:02:00");

Taxonomiska atomer

Paket:

Basis överenskommelser: mestlucid.ephemeral.atomsM SK2taxonomi ( inkluderar MultiTaxonomyAtom)

Atomsorter: mostlylucid.ephemeral.atoms.taxonomy.sensor, .extractor, .embedder, .retriever, .proposer,

.constrainer, .ranker, .renderer, .coordinator, .feedback, .guard

Generiska typer av atomer som hör ihop med taksonomen (sensorn, extraktorn , inbäddaren, , retrievern,, förespråkare,M SK5 konsträngare, bearbetare, koordinator , återkoppling, vakareM SK3 Varje atom släpper ut en utskriven signal när det är färdigt och bär med sig ett Atomkontrakt determinism och persistens metadata. Installera den specifika atompaketet du behöver (exemplet här använder Använda MultiTaxonomyAtom från grundpaketet när du behöver kombinera flera sorter till ett kontrakt.

using Mostlylucid.Ephemeral;
using Mostlylucid.Ephemeral.Atoms.Taxonomy;

public sealed record Proposal(string Text, double Confidence);

var sink = new SignalSink();
var proposer = new ProposerAtom<string, Proposal>(
    sink,
    async (prompt, ct) => new Proposal($"Echo: {prompt}", 0.42),
    outputSignal: "proposal.created",
    keySelector: prompt => prompt);

await proposer.EnqueueAsync("hello");

EscalatorAtom

**Paketet: ** mestlucid.ephemeral.atomsM SK2eskalator

Att förstärka utskrivna ephemerala signaler i hållbara vattensänka. . EscalatorAtomer är det föredrarbara sättet att behålla utgångar från kort-aktiva koordinatorer.

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

public sealed record EscalationPayload(string Kind, string? EvidenceId, double Confidence);

var sink = new SignalSink();
var typed = new TypedSignalSink<EscalationPayload>(sink);

await using var storage = new FileDataStorageAtom<string, EscalationPayload>(
    sink,
    new FileDataStorageConfig { DatabaseName = "signals" });

Func<SignalEvent<EscalationPayload>, CancellationToken, Task> audit = (evt, ct) =>
{
    Console.WriteLine($"audit: {evt.Signal} {evt.Key}");
    return Task.CompletedTask;
};

var targets = new[]
{
    new EscalationTarget<EscalationPayload>(
        "store",
        (evt, ct) => storage.SaveAsync(evt.Key ?? evt.OperationId.ToString(), evt.Payload, ct)),
    new EscalationTarget<EscalationPayload>("audit", audit)
};

await using var escalator = new EscalatorAtom<EscalationPayload>(
    sink,
    typed,
    targets,
    new EscalatorAtomOptions<EscalationPayload>
    {
        EscalateSignalPattern = "escalate.*",
        EmitOnSuccess = "escalation.persisted",
        EmitOnFailure = "escalation.failed"
    });

typed.Raise("escalate.signal", new EscalationPayload("risk", "file-42", 0.81), key: "order-123");

Echo Maker

Paketet: mestlucid.ephemeral.atomsM SK2echo

Capture the typed “last words ” that an operation emitts before it is trimmed. The atom keeps a bounded window of signal lastningar (tillpasande ActivationSignalPattern / CaptureSignalPattern) och när OperationFinalized eld som den producerar OperationEchoEntry<TPayload> rekord du kan behålla genom 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");

Attitydarbeten lyfter bara upp det skrivna signalet med vilken tillstånd de än anser vara kritiskt bounded medan du håller på med echo.

Patterns (Ready-to -UseM SK3

Signalbaserade kretsbinder

**Paketet: ** mestlucid.ephemeral.patternsM SK2circuitbreaker

Stateless strömbrytare med signalhistorie-fenster.

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

SignaldrivenBackpressure

**Paketet: ** mestlucid. efemeral .patronerM SK2 baktryck

Queuedepth management med automatiskt avdrag på baktrycksignaler.

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

Kontrollerad FanOut

**Paketet: ** mestadelslucid. efemeral .patronerM SK2kontrolleradnoutt

Globalt + per-key gating for controlled parallelismM SK2

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

AdaptivRateService

**Paketet: ** mestlucid. efemeral .patronerM SK2adaptivitet

Signal-driven rate limiting with automatic backoff

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

DynamicConcurrencyDemo

**Paketet: ** mestlucid. efemeral .patronerM SK2dynamiskkonkurentitet

Runtime-komparens skalning baserat på dragsignaler.

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

**Paketet: ** mestlucid.ephemeral.patternsM SK2keyedpriorityfanout

Prioritära vägar med per-keyordering bevaradM SK1

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

Reaktiv FanOutPipeline

**Paketet: ** mestadelslucid. efemeral .patroner

Två- trappor med automatiskt baktryck

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

**Paketet: ** mestlucid.ephemeral.patternsM SK2anomalidetektor

Att flytta-fenster-anomali-detektorn.

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

Signalkoordineradeläsningar

**Paketet: ** mestadelslucid. efemeral .patronerM SK2signalkoordinerade läsningar

Quiesce läser under uppdaterade system utan hårdlockar.

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

SignaliseringHttpClient

**Paketet: ** mestlucid.ephemeral.patternsM SK2signaliseringhttp

HTTP-client med framstegsignaler.

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

**Paketet: ** mestlucid.ephemeral.patternsM SK2signallogwatcher

Titta på signalernas venster för mönster och utlösade ringer.

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)

TelemetrisignalHandler

**Paketet: ** mestlucid. efemeral .patronerM SK2telemetri

OpenTelemetry/Integration av applikationsinsikter

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

**Paketet: ** mestlucid.ephemeral.patternsM SK2longwindowdemo

Demonstruerar stor fönstrets konfiguration för auditspår.

using Mostlylucid.Ephemeral.Patterns.LongWindowDemo;

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

Signalreaktion

**Paketet: ** mestadelslucid. efemeral .patroner.signalreaktionsbilden

Demonstruerar signaluppdrag och återkopplingar.

using Mostlylucid.Ephemeral.Patterns.SignalReactionShowcase;

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

"PersstentSignalWindow"

**Paketet: ** mestlucid.ephemeral.patternsM SK2persistentwindow

Signalfenster med SQLite-härdighet - överlever processen återstartar .

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

beroende injection

// 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());
    }
}

Moderna DI-корner kan föredrar kortare hjälper som services.AddCoordinator<T>(...), services.AddScopedCoordinator<T>(...), eller services.AddKeyedCoordinator<T, TKey>(...) eftersom de läser som vanligt AddX Inregistreringar de helt enkelt delegerar till den efemeriska hjälppersonen under kapten

Gadan + Nyckelkonfiguration (Kleisk konfiguration

mostlylucid.ephemeral.complete paket PriorityWorkCoordinator och de keyade varianterna, så samma “Ankoordinator ytan du använder vid uppstart kan skicka arbetet genom namngivna vägar. Konfigurera ett par vägar MaxDepth, och anlita sig på koordinatorn för att dra bort “hotM SK1 ben före “slow” ben medan man fortfarande observerar alla begäranklagar

var sink = new SignalSink();
var lanes = new[]
{
    new PriorityLane("hot:4", CancelOnSignals: new HashSet<string> { "maintenance" }),
    new PriorityLane("normal"),
    new PriorityLane("slow:2")
};

await using var coordinator = new PriorityWorkCoordinator<WorkItem>(
    new PriorityWorkCoordinatorOptions<WorkItem>(
        async (item, ct) => await processor.ProcessAsync(item, ct),
        lanes,
        new EphemeralOptions { Signals = sink }));

await coordinator.EnqueueAsync(new WorkItem("order-42"), laneName: "hot");
await coordinator.EnqueueAsync(new WorkItem("order-43"), laneName: "slow");

Använda PriorityKeyedWorkCoordinator om du också behöver per-", "key ordering", — och "lane decisions" händer fortfarande i pumpen. byggt-in nyckelselektor håller varje partition sequentiell

var keyed = new PriorityKeyedWorkCoordinator<WorkItem, string>(
    new PriorityKeyedWorkCoordinatorOptions<WorkItem, string>(
        order => order.CustomerId,
        async (order, ct) => await processor.ProcessPerCustomerAsync(order, ct),
        lanes,
        new EphemeralOptions { Signals = sink }));

await keyed.EnqueueAsync(new WorkItem("order-42") { CustomerId = "A" }, laneName: "hot");
await keyed.EnqueueAsync(new WorkItem("order-44") { CustomerId = "B" }, laneName: "normal");

Tänk på koordinatorn som kocken, banarna som skålarna, ,, och nyckeln som tallrikar som varje kunds order. sequentiell— den dolda kraften som håller jobbet redo , sprängt, , och signalerna

Raketemang för målen

.NET

Licens

Utanlicenser (public domain)

mest uppenbart. efemeriskt . komplettt; ett märkligt samtida systemmustersystem i en LRU-cache

Priorer

Korn

Attribute & DI

Mostlylucid.Ephemeral.Komplett

Vad är nytt med 2.0

Att bryta mot förändringar

Nya egenskaper

Tabellen med innehåll

Vinnig Start

Registering av tjänster

Attribute-driven jobb

Avsedda uppgifter

Logging & Signaler

Koordinatorer

EphemeralWorkCoordinator<T>

EphemeralKeyedWorkCoordinator<Tkey, T>

Koordinator för tidiga resultat<TInput , resultat>

Prioritetsarbetande koordinator<T>

Konfiguration (EphemeralOptions)

Signaler

ansvarssignaler & Finalisering

Atomer (Blöcke

FixedWorkAtom

KeyedSequentialAtom

SignalAwareAtom

BatchingAtom

Försök om atom

Data lagring atomer

MoleculeRunner & AtomTrigger

SlidingCacheAtom

VolatileOperationAtom

EphemeralLruCache

Atom vid fönstrets storlek

Taxonomiska atomer

EscalatorAtom

Echo Maker

Patterns (Ready-to -UseM SK3

Signalbaserade kretsbinder

SignaldrivenBackpressure

Kontrollerad FanOut

AdaptivRateService

DynamicConcurrencyDemo

KeyedPriorityFanOut

Reaktiv FanOutPipeline

Signalanomalydetektor

Signalkoordineradeläsningar

SignaliseringHttpClient

SignallogWatcher

TelemetrisignalHandler

LongWindowDemo

Signalreaktion

"PersstentSignalWindow"

beroende injection

Gadan + Nyckelkonfiguration (Kleisk konfiguration

Raketemang för målen

Licens