Outbox leases, replay and stale acknowledgments

{
  "sdk": {
    "version": "10.0.403",
    "rollForward": "latestPatch"
  }
}

<Project Sdk="Microsoft.NET.Sdk.Web"><PropertyGroup><TargetFramework>net10.0</TargetFramework><Nullable>enable</Nullable><ImplicitUsings>enable</ImplicitUsings></PropertyGroup><ItemGroup><PackageReference Include="Microsoft.Data.Sqlite" Version="10.0.14"/><PackageReference Include="Microsoft.Extensions.Caching.Hybrid" Version="10.11.0"/></ItemGroup></Project>

using Microsoft.Extensions.Caching.Hybrid;
public sealed record CatalogView(string Name,long Version);
public sealed record CatalogInput(string Name);
public sealed record SharedView(string Value,int Generation);
public sealed class CacheProbe {
    private int reads,factories; private readonly object sync=new(); private TaskCompletionSource<bool> gate=Ready();
    private static TaskCompletionSource<bool> Ready() { var value=new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously); value.SetResult(true); return value; }
    public int Reads=>Volatile.Read(ref reads); public int Factories=>Volatile.Read(ref factories);
    public void Read()=>Interlocked.Increment(ref reads);
    public void Close() { lock(sync) if(gate.Task.IsCompleted) gate=new(TaskCreationOptions.RunContinuationsAsynchronously); }
    public void Open() { lock(sync) gate.TrySetResult(true); }
    public async ValueTask<SharedView> Create(CancellationToken token) { int generation=Interlocked.Increment(ref factories); Task wait; lock(sync) wait=gate.Task; await wait.WaitAsync(token); return new("shared",generation); }
}
public static class CacheLabs {
    public static void Map(WebApplication app) {
        app.MapGet("/cache/catalog",async (Store store,HybridCache cache,CacheProbe probe,CancellationToken token)=> {
            var value=await cache.GetOrCreateAsync("catalog:1",ct=> {
                ct.ThrowIfCancellationRequested(); probe.Read(); using var c=store.Open(); using var query=Store.Command(c,"SELECT name,version FROM catalog_cache WHERE id=1"); using var row=query.ExecuteReader(); row.Read(); return ValueTask.FromResult(new CatalogView(row.GetString(0),row.GetInt64(1)));
            },cancellationToken:token); return Results.Ok(value);
        });
        app.MapPost("/cache/catalog",async (CatalogInput input,bool? failBeforeCommit,Store store,HybridCache cache,CancellationToken token)=> {
            if(string.IsNullOrWhiteSpace(input.Name)||input.Name.Length>30) return Results.BadRequest();
            using(var c=store.Open()) using(var tx=c.BeginTransaction()) {
                Store.Execute(c,"UPDATE catalog_cache SET name=$name,version=version+1 WHERE id=1",tx,("$name",input.Name));
                if(failBeforeCommit==true) { tx.Rollback(); return Results.Problem(statusCode:503,title:"Controlled cache-write rollback"); }
                tx.Commit();
            }
            await cache.RemoveAsync("catalog:1",token); return Results.NoContent();
        });
        app.MapGet("/cache/shared",async (HybridCache cache,CacheProbe probe,CancellationToken token)=>Results.Ok(await cache.GetOrCreateAsync("stampede:demo",ct=>probe.Create(ct),cancellationToken:token)));
        app.MapGet("/cache/stats",(CacheProbe probe)=>new { reads=probe.Reads,factories=probe.Factories });
        app.MapPost("/lab/cache/gate/{action}",(string action,CacheProbe probe)=> { if(action=="close") probe.Close(); else if(action=="open") probe.Open(); else return Results.BadRequest(); return Results.NoContent(); });
        app.MapPost("/lab/cache/reset",async(HybridCache cache,CancellationToken token)=> { await cache.RemoveAsync("stampede:demo",token); return Results.NoContent(); });
    }
}

using System.Text.Json;
public static class ContractLabs {
    public static void Map(WebApplication app) {
        app.MapGet("/contracts/compatible",()=>Results.Ok(new { id=1,name="rune",optional="new" }));
        app.MapGet("/contracts/breaking",()=>Results.Ok(new { id=1,name=42 }));
        app.MapGet("/contracts/effect",(Store store)=> { using var c=store.Open(); return new { value=Store.Number(c,"SELECT value FROM contract_effect WHERE id=1") }; });
        app.MapPost("/contracts/change/{phase}",(string phase,Store store)=> {
            if(phase!="before"&&phase!="after")return Results.BadRequest();
            using var c=store.Open(); using var tx=c.BeginTransaction(); Store.Execute(c,"UPDATE contract_effect SET value=value+1 WHERE id=1",tx);
            if(phase=="before") { tx.Rollback(); return Results.Problem(statusCode:503,title:"Controlled failure before commit"); }
            tx.Commit(); return Results.Problem(statusCode:503,title:"Controlled uncertain response after commit");
        });
    }
    public static bool Accept(JsonElement root) => root.ValueKind==JsonValueKind.Object && root.TryGetProperty("id",out var id)&&id.ValueKind==JsonValueKind.Number&&id.TryGetInt32(out int value)&&value>0&&root.TryGetProperty("name",out var name)&&name.ValueKind==JsonValueKind.String&&!string.IsNullOrWhiteSpace(name.GetString());
    public static async Task VerifyAsync(string address) {
        var uri=new Uri(address); if(!uri.IsLoopback||uri.Scheme!="http"||uri.AbsolutePath!="/"||!string.IsNullOrEmpty(uri.UserInfo))throw new ArgumentException("Owned loopback root required");
        using var client=new HttpClient { BaseAddress=uri,Timeout=TimeSpan.FromSeconds(5) }; int checks=0;
        foreach(var item in new[] { ("contracts/compatible",true),("contracts/breaking",false) }) {
            using var response=await client.GetAsync(item.Item1); response.EnsureSuccessStatusCode(); using var document=JsonDocument.Parse(await response.Content.ReadAsStringAsync());
            if(Accept(document.RootElement)!=item.Item2)throw new Exception("Consumer contract expectation failed"); checks++;
        }
        async Task<long> State() { using var response=await client.GetAsync("contracts/effect"); response.EnsureSuccessStatusCode(); using var document=JsonDocument.Parse(await response.Content.ReadAsStringAsync()); return document.RootElement.GetProperty("value").GetInt64(); }
        if(await State()!=0)throw new Exception("Use a fresh fixture"); checks++;
        using(var response=await client.PostAsync("contracts/change/before",null)) { if((int)response.StatusCode!=503)throw new Exception("Expected failure"); checks++; }
        if(await State()!=0)throw new Exception("Rollback failed"); checks++;
        using(var response=await client.PostAsync("contracts/change/after",null)) { if((int)response.StatusCode!=503)throw new Exception("Expected uncertain response"); checks++; }
        if(await State()!=1)throw new Exception("Commit state missing"); checks++;
        Console.WriteLine(JsonSerializer.Serialize(new { checks,transport="real owned loopback HTTP" }));
    }
}

using Microsoft.Net.Http.Headers;
public sealed record DocumentInput(string Body);
public sealed record DocumentView(int Id,string Body,long Version);
public sealed class ConditionalWriteProbe {
    private readonly object sync=new(); private TaskCompletionSource<bool>? release; private int entered;
    public bool Entered=>Volatile.Read(ref entered)==1;
    public void Arm() { lock(sync) { release=new(TaskCreationOptions.RunContinuationsAsynchronously); Volatile.Write(ref entered,0); } }
    public void Release() { lock(sync) release?.TrySetResult(true); }
    public Task PauseAsync(CancellationToken token) { lock(sync) { if(release is null)return Task.CompletedTask; Volatile.Write(ref entered,1); return release.Task.WaitAsync(token); } }
}
public static class HttpLabs {
    public static DocumentView Read(Store store) { using var c=store.Open(); using var command=Store.Command(c,"SELECT body,version FROM documents WHERE id=1"); using var row=command.ExecuteReader(); row.Read(); return new(1,row.GetString(0),row.GetInt64(1)); }
    public static EntityTagHeaderValue Tag(long version) => new("\"v"+version+"\"");
    public static void Map(WebApplication app) {
        app.MapGet("/documents/1",(HttpContext context,Store store)=> {
            var value=Read(store); var tag=Tag(value.Version); context.Response.GetTypedHeaders().ETag=tag;
            try { var candidates=context.Request.GetTypedHeaders().IfNoneMatch; if(candidates?.Any(candidate=>candidate.Tag.Value=="*" || candidate.Compare(tag,useStrongComparison:false))==true) return Results.StatusCode(304); }
            catch(FormatException) { return Results.BadRequest(new { error="Malformed conditional header" }); }
            return Results.Ok(value);
        });
        app.MapPut("/documents/1",async (HttpContext context,DocumentInput input,Store store,ConditionalWriteProbe probe)=> {
            if(string.IsNullOrWhiteSpace(input.Body)||input.Body.Length>50) return Results.BadRequest();
            IList<EntityTagHeaderValue>? candidates;
            try { candidates=context.Request.GetTypedHeaders().IfMatch; } catch(FormatException) { return Results.BadRequest(); }
            if(candidates is null||candidates.Count==0) return Results.Problem(statusCode:428,title:"If-Match is required");
            if(context.Request.Headers["X-Lab-Pause"]=="true")await probe.PauseAsync(context.RequestAborted);
            using var c=store.Open(); using var tx=c.BeginTransaction();
            long current=Store.Number(c,"SELECT version FROM documents WHERE id=1",tx);
            bool any=candidates.Any(x=>x.Tag.Value=="*");
            if(!any&&!candidates.Any(x=>x.Compare(Tag(current),useStrongComparison:true))) { tx.Rollback(); return Results.StatusCode(412); }
            using var update=Store.Command(c,"UPDATE documents SET body=$body,version=version+1 WHERE id=1 AND version=$version RETURNING version",tx,("$body",input.Body),("$version",current));
            object? changed=update.ExecuteScalar(); if(changed is null) { tx.Rollback(); return Results.StatusCode(412); }
            long version=Convert.ToInt64(changed,System.Globalization.CultureInfo.InvariantCulture); tx.Commit(); context.Response.GetTypedHeaders().ETag=Tag(version); return Results.Ok(new DocumentView(1,input.Body,version));
        });
        app.MapPost("/lab/condition/arm",(ConditionalWriteProbe probe)=> { probe.Arm(); return Results.NoContent(); });
        app.MapPost("/lab/condition/release",(ConditionalWriteProbe probe)=> { probe.Release(); return Results.NoContent(); });
        app.MapGet("/lab/condition/evidence",(ConditionalWriteProbe probe)=>new { entered=probe.Entered });
        app.MapGet("/api/v1/item",()=>Results.Ok(new { id=1,name="rune" }));
        app.MapGet("/api/v1/item-extra",()=>Results.Ok(new { id=1,name="rune",state="active" }));
        app.MapGet("/api/v2/item",()=>Results.Ok(new { id=1,displayName="rune",state="active" }));
        app.MapGet("/stream",async (HttpContext context,int? abortAfter,bool? fail)=> {
            if(abortAfter is not null&&abortAfter!=1) { context.Response.StatusCode=400; return; }
            context.Response.ContentType="application/x-ndjson";
            for(int id=1;id<=2;id++) {
                await context.Response.WriteAsync(System.Text.Json.JsonSerializer.Serialize(new { type="item",id })+"\n",context.RequestAborted); await context.Response.Body.FlushAsync(context.RequestAborted);
                if(abortAfter==id) { context.Abort(); return; }
                if(fail==true&&id==1) { await context.Response.WriteAsync("{\"type\":\"error\",\"code\":\"controlled-failure\"}\n",context.RequestAborted); return; }
            }
            await context.Response.WriteAsync("{\"type\":\"complete\",\"count\":2}\n",context.RequestAborted);
        });
    }
}

using System.Text.Json;
using Microsoft.Data.Sqlite;
public sealed record CreateOrder(string OrderId,int Quantity);
public sealed record InboxInput(string MessageId,string Payload,int Amount);
public sealed record OrderedInput(string MessageId,string AggregateId,int Sequence,int Amount);
public static class MessagingLabs {
    public static void Map(WebApplication app) {
        app.MapGet("/messaging/counts",(Store store)=>store.Counts());
        app.MapPost("/outbox/orders", (CreateOrder input, bool? failBeforeCommit, Store store) => {
            if(input.OrderId!="demo" || input.Quantity is <1 or >10) return Results.BadRequest();
            using var c=store.Open(); using var tx=c.BeginTransaction();
            try {
            Store.Execute(c,"INSERT INTO orders(order_id,quantity) VALUES($id,$quantity)",tx,("$id",input.OrderId),("$quantity",input.Quantity));
            Store.Execute(c,"INSERT INTO outbox(event_key,aggregate_id,sequence,payload) VALUES($key,$id,1,$payload)",tx,("$key","order-demo-v1"),("$id",input.OrderId),("$payload","quantity:"+input.Quantity));
            } catch(SqliteException ex) when(ex.SqliteErrorCode==19) { tx.Rollback(); return Results.Conflict(new { error="Order or event identity already exists" }); }
            if(failBeforeCommit==true) { tx.Rollback(); return Results.Problem(statusCode:503,title:"Controlled pre-commit failure"); }
            tx.Commit(); return Results.Created("/outbox/orders/demo",new { input.OrderId,input.Quantity });
        });
        app.MapPost("/outbox/claim/{owner}",(string owner,Store store)=> {
            if(owner!="one"&&owner!="two") return Results.BadRequest(); long now=DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
            using var c=store.Open(); using var tx=c.BeginTransaction();
            using var command=Store.Command(c,"UPDATE outbox SET owner=$owner,lease_until=$until,attempts=attempts+1 WHERE id=(SELECT id FROM outbox WHERE status='pending' AND (lease_until IS NULL OR lease_until<=$now) ORDER BY id LIMIT 1) RETURNING id,payload,attempts",tx,("$owner",owner),("$until",now+500),("$now",now));
            object? row=null; using(var read=command.ExecuteReader()) if(read.Read()) row=new { id=read.GetInt64(0),payload=read.GetString(1),attempts=read.GetInt32(2),owner };
            tx.Commit(); return row is null ? Results.NoContent() : Results.Ok(row);
        });
        app.MapPost("/outbox/ack/{id:long}/{owner}",(long id,string owner,int generation,Store store)=> {
            using var c=store.Open(); int changed=Store.Execute(c,"UPDATE outbox SET status='sent' WHERE id=$id AND owner=$owner AND status='pending' AND lease_until>$now AND attempts=$generation",null,("$id",id),("$owner",owner),("$now",DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()),("$generation",generation)); return changed==1 ? Results.NoContent() : Results.Conflict(new { error="Lease is not currently owned" });
        });
        app.MapPost("/inbox/apply",(InboxInput input,bool? failBeforeCommit,Store store)=> {
            if(input.MessageId!="demo" || (string.IsNullOrWhiteSpace(input.Payload)||input.Payload.Length>50) || input.Amount is <1 or >10) return Results.BadRequest();
            string canonical=JsonSerializer.Serialize(new { input.Payload,input.Amount });
            using var c=store.Open(); using var tx=c.BeginTransaction(); using var check=Store.Command(c,"SELECT payload FROM inbox WHERE message_id=$id",tx,("$id",input.MessageId)); string? prior=check.ExecuteScalar() as string;
            if(prior is not null) { tx.Rollback(); return prior==canonical ? Results.Ok(new { replay=true }) : Results.Conflict(new { error="Payload conflict" }); }
            Store.Execute(c,"INSERT INTO inbox(message_id,payload) VALUES($id,$payload)",tx,("$id",input.MessageId),("$payload",canonical)); Store.Execute(c,"UPDATE effects SET total=total+$amount WHERE id=1",tx,("$amount",input.Amount));
            if(failBeforeCommit==true) { tx.Rollback(); return Results.Problem(statusCode:503,title:"Controlled inbox failure"); } tx.Commit(); return Results.Ok(new { replay=false });
        });
        app.MapPost("/ordered/apply",(OrderedInput input,Store store)=> {
            if((string.IsNullOrWhiteSpace(input.MessageId)||input.MessageId.Length>20) || input.AggregateId!="demo" || input.Sequence is <1 or >10) return Results.BadRequest();
            using var c=store.Open(); using var tx=c.BeginTransaction();
            string canonical=JsonSerializer.Serialize(new { input.AggregateId,input.Sequence,input.Amount });
            using var priorCommand=Store.Command(c,"SELECT payload FROM ordered_inbox WHERE message_id=$id",tx,("$id",input.MessageId)); string? prior=priorCommand.ExecuteScalar() as string;
            if(prior is not null) { tx.Rollback(); return prior==canonical ? Results.Ok(new { outcome="replay" }) : Results.Conflict(new { outcome="payload conflict" }); }
            Store.Execute(c,"INSERT OR IGNORE INTO aggregate_state VALUES($id,0,0)",tx,("$id",input.AggregateId)); long last=Store.Number(c,"SELECT last_sequence FROM aggregate_state WHERE aggregate_id=$id",tx,("$id",input.AggregateId));
            if(input.Sequence!=last+1) { tx.Rollback(); return Results.Conflict(new { outcome=input.Sequence>last+1 ? "gap" : "stale",expected=last+1 }); }
            if(input.Amount<0) { Store.Execute(c,"INSERT OR IGNORE INTO deadletters VALUES($id,'negative amount')",tx,("$id",input.MessageId)); tx.Commit(); return Results.UnprocessableEntity(new { outcome="quarantined",expected=last+1 }); }
            Store.Execute(c,"INSERT INTO ordered_inbox VALUES($id,$payload)",tx,("$id",input.MessageId),("$payload",canonical)); Store.Execute(c,"UPDATE aggregate_state SET last_sequence=$sequence,total=total+$amount WHERE aggregate_id=$id",tx,("$sequence",input.Sequence),("$amount",input.Amount),("$id",input.AggregateId)); tx.Commit(); return Results.Ok(new { outcome="applied",sequence=input.Sequence });
        });
        app.MapGet("/ordered/state",(Store store)=> { using var c=store.Open(); return new { applied=Store.Number(c,"SELECT COUNT(*) FROM ordered_inbox"),deadletters=Store.Number(c,"SELECT COUNT(*) FROM deadletters"),last=Store.Number(c,"SELECT COALESCE(MAX(last_sequence),0) FROM aggregate_state") }; });
    }
}

using Microsoft.Net.Http.Headers;
using System.Text.Json;
public sealed record PipelineInput(int Quantity);
public static class PipelineLabs {
    public static void Map(WebApplication app) {
        app.MapPost("/pipeline",(HttpContext context,PipelineInput input,bool? failAfterCommit,Store store)=> {
            if(input.Quantity is <1 or >10)return Results.BadRequest();
            string requestId=context.Request.Headers["Idempotency-Key"].ToString(); if(requestId is not "demo" and not "second")return Results.BadRequest();
            IList<EntityTagHeaderValue>? candidates;
            try { candidates=context.Request.GetTypedHeaders().IfMatch; } catch(FormatException) { return Results.BadRequest(); }
            if(candidates is null||candidates.Count==0)return Results.StatusCode(428);
            string originalCondition=string.Join(",",candidates.Select(x=>x.ToString()));
            string payload=JsonSerializer.Serialize(new { input.Quantity,condition=originalCondition });
            using var c=store.Open(); using var tx=c.BeginTransaction();
            using(var query=Store.Command(c,"SELECT payload,result_version FROM pipeline_requests WHERE request_id=$id",tx,("$id",requestId))) using(var read=query.ExecuteReader()) {
                if(read.Read()) {
                    string prior=read.GetString(0); long version=read.GetInt64(1); read.Close();
                    long currentVersion=Store.Number(c,"SELECT version FROM documents WHERE id=1",tx); tx.Rollback();
                    return prior==payload&&currentVersion==version ? Results.Ok(new { version,replay=true }) : Results.Conflict(new { error="Request changed or the saved receipt is no longer current" });
                }
            }
            long current=Store.Number(c,"SELECT version FROM documents WHERE id=1",tx);
            if(!candidates.Any(x=>x.Tag.Value=="*"||x.Compare(HttpLabs.Tag(current),useStrongComparison:true))) { tx.Rollback(); return Results.StatusCode(412); }
            long next=current+1; Store.Execute(c,"UPDATE documents SET body=$body,version=$next WHERE id=1 AND version=$current",tx,("$body","quantity:"+input.Quantity),("$next",next),("$current",current));
            Store.Execute(c,"INSERT INTO pipeline_requests VALUES($id,$payload,$version)",tx,("$id",requestId),("$payload",payload),("$version",next));
            Store.Execute(c,"INSERT INTO outbox(event_key,aggregate_id,sequence,payload) VALUES($key,'document-1',$sequence,$payload)",tx,("$key","document-1-v"+next),("$sequence",next),("$payload",payload)); tx.Commit();
            if(failAfterCommit==true)return Results.Problem(statusCode:503,title:"Controlled lost-success response after commit");
            context.Response.Headers.ETag="\"v"+next+"\""; return Results.Ok(new { version=next,replay=false });
        });
        app.MapGet("/pipeline/state",(Store store)=> { using var c=store.Open(); return new { version=Store.Number(c,"SELECT version FROM documents WHERE id=1"),requests=Store.Number(c,"SELECT COUNT(*) FROM pipeline_requests"),events=Store.Number(c,"SELECT COUNT(*) FROM outbox") }; });
    }
}

using System.Diagnostics.Metrics;
public sealed class EvidenceLog : ILoggerProvider {
    private readonly object sync=new(); private readonly Queue<string> messages=new();
    public string[] Snapshot() { lock(sync) return messages.ToArray(); }
    public ILogger CreateLogger(string categoryName)=>new Writer(this,categoryName);
    public void Dispose() { }
    private void Add(string category,string text) { if(category=="ProductionPrivacy") lock(sync) { if(messages.Count==32) messages.Dequeue(); messages.Enqueue(text); } Console.WriteLine(text); }
    private sealed class Writer(EvidenceLog owner,string category):ILogger {
        public IDisposable? BeginScope<TState>(TState state) where TState:notnull=>null;
        public bool IsEnabled(LogLevel level)=>category=="Microsoft.Hosting.Lifetime"||category=="ProductionPrivacy";
        public void Log<TState>(LogLevel level,EventId id,TState state,Exception? exception,Func<TState,Exception?,string> formatter) { if(IsEnabled(level)) owner.Add(category,formatter(state,exception)); }
    }
}
public sealed class PrivacySignals : IDisposable {
    private readonly Meter meter=new("SharpRune.ProductionPrivacy"); private readonly MeterListener listener=new(); private readonly Counter<long> counter;
    private readonly object sync=new(); private readonly HashSet<string> labels=new(StringComparer.Ordinal); private int disposed;
    public PrivacySignals() {
        counter=meter.CreateCounter<long>("production.operations");
        listener.InstrumentPublished=(instrument,value)=> { if(instrument.Meter.Name=="SharpRune.ProductionPrivacy") value.EnableMeasurementEvents(instrument); };
        listener.SetMeasurementEventCallback<long>((instrument,value,tags,state)=> { foreach(var tag in tags) if(tag.Key=="operation"&&tag.Value is string label) lock(sync) labels.Add(label); }); listener.Start();
    }
    public string Record(string raw) { string operation=raw is "quote" or "search" ? raw : "other"; counter.Add(1,new KeyValuePair<string,object?>("operation",operation)); return operation; }
    public string[] Labels() { lock(sync) return labels.OrderBy(x=>x,StringComparer.Ordinal).ToArray(); }
    public void Dispose() { if(Interlocked.Exchange(ref disposed,1)!=0)return; listener.Dispose(); meter.Dispose(); }
}

using System.Net;
using Microsoft.AspNetCore.DataProtection;
using Microsoft.AspNetCore.HttpOverrides;
using Microsoft.Extensions.Caching.Hybrid;
if(args.Length==2&&args[0]=="--verify-base") { await ContractLabs.VerifyAsync(args[1]); return; }
var builder=WebApplication.CreateBuilder(args);
if(!builder.Environment.IsDevelopment())throw new InvalidOperationException("This fixture exposes controlled faults and is Development-only");
var evidence=new EvidenceLog(); builder.Logging.ClearProviders(); builder.Logging.AddProvider(evidence);
builder.Configuration.AddJsonFile("runtime-settings.json",optional:false,reloadOnChange:true);
builder.Services.AddOptions<LiveSettings>().Bind(builder.Configuration.GetSection("Live")).Validate(x=>!string.IsNullOrWhiteSpace(x.Label)&&x.Maximum is >=1 and <=10,"Invalid live settings").ValidateOnStart();
builder.Services.AddSingleton<ConditionalWriteProbe>(); builder.Services.AddSingleton<Store>(); builder.Services.AddSingleton<CacheProbe>(); builder.Services.AddSingleton<PrivacySignals>();
builder.Services.AddHybridCache(options=>options.DefaultEntryOptions=new HybridCacheEntryOptions { Expiration=TimeSpan.FromMinutes(1),LocalCacheExpiration=TimeSpan.FromMinutes(1) });
builder.Services.AddDataProtection().UseEphemeralDataProtectionProvider();
builder.Services.AddAntiforgery(options=> { options.HeaderName="X-CSRF-TOKEN"; options.Cookie.Name="ProductionLab.Csrf"; options.Cookie.SecurePolicy=CookieSecurePolicy.None; });
builder.Services.AddCors(options=>options.AddPolicy("classroom",policy=>policy.WithOrigins("https://client.example").WithMethods("GET","POST").WithHeaders("Content-Type")));
builder.Services.Configure<ForwardedHeadersOptions>(options=> {
    options.ForwardedHeaders=ForwardedHeaders.XForwardedFor|ForwardedHeaders.XForwardedProto; options.ForwardLimit=1;
    options.KnownProxies.Clear(); options.KnownIPNetworks.Clear(); options.KnownProxies.Add(IPAddress.Parse(builder.Configuration["Production:Proxy"]??"203.0.113.10"));
});
builder.Services.AddHostFiltering(options=>options.AllowedHosts=(builder.Configuration["AllowedHosts"]??"*").Split(';'));
builder.Services.AddSingleton<WorkQueue>(); builder.Services.AddSingleton<ScopeEvidence>(); builder.Services.AddScoped<ScopedOperation>(); builder.Services.AddHostedService<ScopedWorker>();
var app=builder.Build(); app.Services.GetRequiredService<Store>().Initialize();
app.UseHostFiltering(); app.UseForwardedHeaders(); app.UseCors("classroom"); app.UseAntiforgery();
SqlLabs.Map(app); MessagingLabs.Map(app); HttpLabs.Map(app); CacheLabs.Map(app); SecurityLabs.Map(app,evidence); WorkerLabs.Map(app); ContractLabs.Map(app); PipelineLabs.Map(app);
app.MapGet("/live",()=>Results.Text("live")); app.Run();

using System.Text.Encodings.Web;
using Microsoft.AspNetCore.Antiforgery;
using Microsoft.Extensions.Options;
public sealed class LiveSettings { public string Label { get; set; }="initial"; public int Maximum { get; set; }=3; public string Secret { get; set; }=""; }
public static class SecurityLabs {
    private static int submissions;
    public static void Map(WebApplication app,EvidenceLog evidence) {
        app.MapGet("/configuration",(IOptionsMonitor<LiveSettings> monitor)=> {
            try { var value=monitor.CurrentValue; return Results.Ok(new { value.Label,value.Maximum,hasSecret=!string.IsNullOrEmpty(value.Secret) }); }
            catch(OptionsValidationException) { return Results.Problem(statusCode:503,title:"Current configuration is invalid"); }
        });
        app.MapGet("/cors/data",(HttpContext context)=>Results.Ok(new { value="public classroom data",authenticated=context.User.Identity?.IsAuthenticated??false }));
        app.MapGet("/antiforgery/token",(HttpContext context,IAntiforgery anti)=> { var tokens=anti.GetAndStoreTokens(context); return Results.Ok(new { requestToken=tokens.RequestToken }); });
        app.MapPost("/antiforgery/submit",async(HttpContext context,IAntiforgery anti)=> {
            try { await anti.ValidateRequestAsync(context); }
            catch(AntiforgeryValidationException) { return Results.BadRequest(new { error="Antiforgery validation failed" }); }
            return Results.Ok(new { accepted=Interlocked.Increment(ref submissions) });
        });
        app.MapGet("/antiforgery/count",()=>new { count=Volatile.Read(ref submissions) });
        app.MapGet("/encoded",(string? name)=>Results.Content("<!doctype html><html><body><p>"+HtmlEncoder.Default.Encode(name??"")+"</p></body></html>","text/html; charset=utf-8"));
        app.MapGet("/proxy",(HttpContext context)=>new { scheme=context.Request.Scheme,remote=context.Connection.RemoteIpAddress?.ToString(),host=context.Request.Host.Value });
        app.MapGet("/canonical",(HttpContext context)=>new { observedHost=context.Request.Host.Value,url=new Uri(new Uri("https://public.example.test/"),"items/7").AbsoluteUri });
        app.MapGet("/privacy/observe",(string? email,string? operation,PrivacySignals signals,ILoggerFactory logs)=> {
            string safe=signals.Record(operation??""); string identity=string.IsNullOrEmpty(email)?"anonymous":"redacted";
            logs.CreateLogger("ProductionPrivacy").LogInformation("Handled {Operation} for {Identity}",safe,identity); return Results.Ok(new { operation=safe,identity });
        });
        app.MapGet("/privacy/evidence",(PrivacySignals signals)=>new { messages=evidence.Snapshot(),metricLabels=signals.Labels() });
    }
}

using Microsoft.Data.Sqlite;
public static class SqlLabs {
    public static void Map(WebApplication app) {
        app.MapGet("/sql/nulls", (Store store) => { using var c = store.Open(); return new { equalsNull = Store.Number(c, "SELECT COUNT(*) FROM sql_values WHERE amount = NULL"), isNull = Store.Number(c, "SELECT COUNT(*) FROM sql_values WHERE amount IS NULL"), all = Store.Number(c, "SELECT COUNT(*) FROM sql_values"), known = Store.Number(c, "SELECT COUNT(amount) FROM sql_values"), sum = Store.Number(c, "SELECT SUM(amount) FROM sql_values") }; });
        app.MapPost("/lab/sql/null-check", (Store store) => {
            using var c = store.Open(); using var tx = c.BeginTransaction(); Store.Execute(c,"INSERT INTO sql_values(id,amount) VALUES(6,NULL)",tx);
            bool rejected = false; try { Store.Execute(c,"INSERT INTO sql_required(amount) VALUES(NULL)",tx); } catch(SqliteException ex) when(ex.SqliteErrorCode==19) { rejected=true; }
            tx.Rollback(); return new { checkAllowedNull = true, notNullRejected = rejected };
        });
        app.MapGet("/sql/plans", (Store store) => {
            using var c = store.Open();
            var rows = new List<int>(); using(var command=Store.Command(c,"SELECT id,price FROM products WHERE category=$category ORDER BY price,id",null,("$category","books"))) using(var read=command.ExecuteReader()) while(read.Read()) rows.Add(read.GetInt32(0));
            return new { ids=rows, composite=Plan(c,"SELECT id,price FROM products WHERE category='books' ORDER BY price,id"), expression=Plan(c,"SELECT id FROM products WHERE lower(name)='rune'"), collated=Plan(c,"SELECT id FROM products WHERE name='rune' COLLATE NOCASE"), partial=Plan(c,"SELECT id,price FROM products WHERE active=1 ORDER BY price") };
        });
        app.MapPost("/lab/sql/snapshot", (Store store) => {
            using var first=store.Open(); Store.Execute(first,"UPDATE counters SET value=10 WHERE id=1"); using var reader=first.BeginTransaction(deferred:true);
            long before=Store.Number(first,"SELECT value FROM counters WHERE id=1",reader);
            using(var second=store.Open()) Store.Execute(second,"UPDATE counters SET value=20 WHERE id=1");
            long during=Store.Number(first,"SELECT value FROM counters WHERE id=1",reader); reader.Commit();
            using var fresh=store.Open(); return new { before,during,after=Store.Number(fresh,"SELECT value FROM counters WHERE id=1") };
        });
        app.MapPost("/lab/sql/retry", (Store store) => {
            using var first=store.Open(); Store.Execute(first,"UPDATE counters SET value=10 WHERE id=1"); using var stale=first.BeginTransaction(deferred:true);
            long seen=Store.Number(first,"SELECT value FROM counters WHERE id=1",stale);
            using(var second=store.Open()) Store.Execute(second,"UPDATE counters SET value=20 WHERE id=1");
            int error=0,extended=0; try { Store.Execute(first,"UPDATE counters SET value=$value WHERE id=1",stale,("$value",seen+1)); } catch(SqliteException ex) when(ex.SqliteErrorCode==5 || ex.SqliteErrorCode==6) { error=ex.SqliteErrorCode; extended=ex.SqliteExtendedErrorCode; }
            stale.Rollback(); if(error==0) return Results.Problem(statusCode:500,title:"Fixture did not encounter the expected stale-writer boundary");
            using var fresh=store.Open(); using var tx=fresh.BeginTransaction(); long current=Store.Number(fresh,"SELECT value FROM counters WHERE id=1",tx); Store.Execute(fresh,"UPDATE counters SET value=$value WHERE id=1",tx,("$value",current+1)); tx.Commit();
            return Results.Ok(new { attempts=2,error,extended,value=current+1 });
        });
    }
    private static List<string> Plan(SqliteConnection c,string sql) { var rows=new List<string>(); using var command=Store.Command(c,"EXPLAIN QUERY PLAN "+sql); using var read=command.ExecuteReader(); while(read.Read()) rows.Add(read.GetString(3)); return rows; }
}

using Microsoft.Data.Sqlite;
public sealed class Store {
    private readonly string connectionString;
    public Store(IConfiguration configuration) {
        string path = configuration["Production:DatabasePath"] ?? Path.Combine(Path.GetTempPath(), "sharprune-production-" + Guid.NewGuid().ToString("N") + ".db");
        connectionString = new SqliteConnectionStringBuilder { DataSource = path, DefaultTimeout = 1, Pooling = false, ForeignKeys = true }.ToString();
    }
    public SqliteConnection Open() { var connection = new SqliteConnection(connectionString); connection.Open(); return connection; }
    public static SqliteCommand Command(SqliteConnection c, string sql, SqliteTransaction? tx = null, params (string Key, object? Value)[] values) {
        var command = c.CreateCommand(); command.CommandText = sql; command.Transaction = tx;
        foreach (var value in values) command.Parameters.AddWithValue(value.Key, value.Value ?? DBNull.Value);
        return command;
    }
    public static int Execute(SqliteConnection c, string sql, SqliteTransaction? tx = null, params (string Key, object? Value)[] values) { using var command = Command(c, sql, tx, values); return command.ExecuteNonQuery(); }
    public static long Number(SqliteConnection c, string sql, SqliteTransaction? tx = null, params (string Key, object? Value)[] values) { using var command = Command(c, sql, tx, values); return Convert.ToInt64(command.ExecuteScalar(), System.Globalization.CultureInfo.InvariantCulture); }
    public void Initialize() {
        using var c = Open(); Execute(c, "PRAGMA journal_mode=WAL");
        Execute(c, """
CREATE TABLE IF NOT EXISTS sql_values(id INTEGER PRIMARY KEY, amount INTEGER CHECK(amount > 0));
CREATE TABLE IF NOT EXISTS sql_required(amount INTEGER NOT NULL CHECK(amount > 0));
INSERT OR IGNORE INTO sql_values VALUES(1,NULL),(2,NULL),(3,1),(4,2),(5,3);
CREATE TABLE IF NOT EXISTS products(id INTEGER PRIMARY KEY,name TEXT NOT NULL,category TEXT NOT NULL,price INTEGER NOT NULL CHECK(price>=0),active INTEGER NOT NULL CHECK(active IN(0,1)));
INSERT OR IGNORE INTO products VALUES(1,'Rune','books',30,1),(2,'rune','books',10,1),(3,'Sword','books',20,0),(4,'Shield','gear',20,1),(5,'RUNE','gear',15,0);
CREATE INDEX IF NOT EXISTS ix_products_category_price_id ON products(category,price,id);
CREATE INDEX IF NOT EXISTS ix_products_active_price ON products(price) WHERE active=1;
CREATE INDEX IF NOT EXISTS ix_products_lower_name ON products(lower(name));
CREATE TABLE IF NOT EXISTS counters(id INTEGER PRIMARY KEY,value INTEGER NOT NULL);
INSERT OR IGNORE INTO counters VALUES(1,10);
CREATE TABLE IF NOT EXISTS orders(order_id TEXT PRIMARY KEY,quantity INTEGER NOT NULL CHECK(quantity BETWEEN 1 AND 10));
CREATE TABLE IF NOT EXISTS outbox(id INTEGER PRIMARY KEY AUTOINCREMENT,event_key TEXT UNIQUE NOT NULL,aggregate_id TEXT NOT NULL,sequence INTEGER NOT NULL,payload TEXT NOT NULL,status TEXT NOT NULL DEFAULT 'pending',owner TEXT,lease_until INTEGER,attempts INTEGER NOT NULL DEFAULT 0);
CREATE TABLE IF NOT EXISTS inbox(message_id TEXT PRIMARY KEY,payload TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS effects(id INTEGER PRIMARY KEY,total INTEGER NOT NULL);
INSERT OR IGNORE INTO effects VALUES(1,0);
CREATE TABLE IF NOT EXISTS aggregate_state(aggregate_id TEXT PRIMARY KEY,last_sequence INTEGER NOT NULL,total INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS ordered_inbox(message_id TEXT PRIMARY KEY,payload TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS deadletters(message_id TEXT PRIMARY KEY,reason TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS documents(id INTEGER PRIMARY KEY,body TEXT NOT NULL,version INTEGER NOT NULL);
INSERT OR IGNORE INTO documents VALUES(1,'rune',1);
CREATE TABLE IF NOT EXISTS catalog_cache(id INTEGER PRIMARY KEY,name TEXT NOT NULL,version INTEGER NOT NULL);
INSERT OR IGNORE INTO catalog_cache VALUES(1,'rune',1);
CREATE TABLE IF NOT EXISTS contract_effect(id INTEGER PRIMARY KEY,value INTEGER NOT NULL);
INSERT OR IGNORE INTO contract_effect VALUES(1,0);
CREATE TABLE IF NOT EXISTS pipeline_requests(request_id TEXT PRIMARY KEY,payload TEXT NOT NULL,result_version INTEGER NOT NULL);
""");
    }
    public object Counts() { using var c = Open(); return new { orders = Number(c, "SELECT COUNT(*) FROM orders"), outbox = Number(c, "SELECT COUNT(*) FROM outbox"), inbox = Number(c, "SELECT COUNT(*) FROM inbox"), effects = Number(c, "SELECT total FROM effects WHERE id=1") }; }
}

using System.Threading.Channels;
public sealed record WorkRequest(int Value);
public sealed record WorkReceipt(int Value,int ScopeId);
public sealed record WorkMessage(int Value,CancellationToken Token) { public TaskCompletionSource<WorkReceipt> Completion { get; }=new(TaskCreationOptions.RunContinuationsAsynchronously); }
public sealed class WorkQueue {
    public Channel<WorkMessage> Items { get; }=Channel.CreateBounded<WorkMessage>(4);
}
public sealed class ScopeEvidence {
    private int created,disposed,live;
    public int Created=>Volatile.Read(ref created); public int Disposed=>Volatile.Read(ref disposed); public int Live=>Volatile.Read(ref live);
    public int Start() { Interlocked.Increment(ref live); return Interlocked.Increment(ref created); }
    public void End() { Interlocked.Decrement(ref live); Interlocked.Increment(ref disposed); }
}
public sealed class ScopedOperation : IAsyncDisposable {
    private readonly ScopeEvidence evidence; private int disposed; public int Id { get; }
    public ScopedOperation(ScopeEvidence evidence) { this.evidence=evidence; Id=evidence.Start(); }
    public async Task<WorkReceipt> ProcessAsync(int value,CancellationToken token) { await Task.Delay(2,token); return new(value*2,Id); }
    public ValueTask DisposeAsync() { if(Interlocked.Exchange(ref disposed,1)==0)evidence.End(); return ValueTask.CompletedTask; }
}
public sealed class ScopedWorker(WorkQueue queue,IServiceScopeFactory scopes) : BackgroundService {
    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
        await foreach(var item in queue.Items.Reader.ReadAllAsync(stoppingToken)) {
            using var owner=CancellationTokenSource.CreateLinkedTokenSource(stoppingToken,item.Token);
            try {
                WorkReceipt result;
                await using(var scope=scopes.CreateAsyncScope()) result=await scope.ServiceProvider.GetRequiredService<ScopedOperation>().ProcessAsync(item.Value,owner.Token);
                item.Completion.TrySetResult(result);
            } catch(OperationCanceledException) { item.Completion.TrySetCanceled(owner.Token); }
        }
    }
    public override async Task StopAsync(CancellationToken token) {
        queue.Items.Writer.TryComplete();
        try { if(ExecuteTask is not null) await ExecuteTask.WaitAsync(token); }
        finally { await base.StopAsync(token); }
    }
}
public static class WorkerLabs {
    public static void Map(WebApplication app) {
        app.MapPost("/worker",async(WorkRequest input,WorkQueue queue,CancellationToken token)=> {
            if(input.Value is <1 or >10)return Results.BadRequest(); var item=new WorkMessage(input.Value,token);
            if(!queue.Items.Writer.TryWrite(item))return Results.StatusCode(429);
            return Results.Ok(await item.Completion.Task.WaitAsync(token));
        });
        app.MapGet("/worker/evidence",(ScopeEvidence evidence)=>new { evidence.Created,evidence.Disposed,evidence.Live });
    }
}

{
  "AllowedHosts": "*",
  "Production": {
    "Proxy": "203.0.113.10"
  }
}

{
  "Live": {
    "Label": "initial",
    "Maximum": 3,
    "Secret": "classroom-not-a-secret"
  }
}