diff --git a/cmd/agentop/edit/edit.go b/cmd/agentop/edit/edit.go index 42e0b0be3..1bd005326 100644 --- a/cmd/agentop/edit/edit.go +++ b/cmd/agentop/edit/edit.go @@ -174,13 +174,9 @@ type PolledMsg struct { } // ClusterPollDeadline bounds a ConfigMap edit's wait: the worst-case kubelet -// sync (~60s) plus the framework's drain window (30s) plus jitter. -// -// The drain does NOT actually delay the verdict — the reloader records success -// synchronously right after the Holder swap and drains the old pipelines in a -// goroutine — so this is generous rather than tight. Kept at 120s regardless, -// because for a cluster the kubelet sync is the real variable and it is not -// worth shaving. +// sync (~60s) plus jitter, doubled. The verdict does not wait on the drain — +// the reloader records success right after the Holder swap and drains the old +// pipelines in a goroutine. const ClusterPollDeadline = 120 * time.Second // LocalPollDeadline bounds a local file edit's wait. The proxy is already diff --git a/core/listener/forwardproxy/websocket.go b/core/listener/forwardproxy/websocket.go index 50fc3aad8..5be7b9f75 100644 --- a/core/listener/forwardproxy/websocket.go +++ b/core/listener/forwardproxy/websocket.go @@ -329,11 +329,9 @@ func (ws *wsSocket) wsEnd(side string) { type wsTurn struct { pctx *pipeline.Context // p is the pipeline the turn began on, and every later step of the turn runs on - // it. A hot reload swaps OutboundPipeline for new requests and keeps the old one - // running for its drain window so requests already in flight finish where they - // started; an HTTPS turn does that by lasting seconds. A socket outlives the - // window, so a turn looking the pipeline up again at each step would end on one - // whose parser never saw it begin — no tokens, no cost. + // it, and the turn holds it until its Finishers run. A hot reload swaps + // OutboundPipeline for new requests, so a turn looking the pipeline up again at + // each step would end on one whose parser never saw it begin — no tokens, no cost. p *pipeline.Pipeline // prewarm is a generate:false response.create: run through the request phase, not // recorded, and its answer not parsed. @@ -611,17 +609,18 @@ func (ws *wsSocket) clientMessage(payload []byte) (replace []byte, refuse bool) } body, generate := responseCreateBody(payload) - p := ws.s.OutboundPipeline.Load() + pctx := ws.newTurnContext(body) + p := ws.s.OutboundPipeline.Pin(pctx) if why := pipelineCannotCarry(p); why != "" { // A reload since the handshake made this socket one the proxy would have // declined. End it with "try again later": Codex reconnects, and the new // handshake is answered 426 and falls back to HTTPS, where the plugin applies. slog.Info("forward-proxy: ending a carried WebSocket after a reload", "host", ws.r.Host, "reason", why) + p.RunFinish(ws.ctx(), pctx, pipeline.Outcome{}) ws.closeBoth(wsframe.CloseTryAgainLater, "the proxy's pipeline changed; reconnect") return nil, true } - pctx := ws.newTurnContext(body) t := &wsTurn{pctx: pctx, p: p, prewarm: !generate} action := p.Run(ws.ctx(), pctx) if action.Type == pipeline.Reject { diff --git a/core/listener/forwardproxy/websocket_review_test.go b/core/listener/forwardproxy/websocket_review_test.go index 48bcf472f..0290492d1 100644 --- a/core/listener/forwardproxy/websocket_review_test.go +++ b/core/listener/forwardproxy/websocket_review_test.go @@ -155,6 +155,9 @@ func TestResponsesWebSocket_ReloadAddingAnUncarriablePluginEndsTheSocket(t *test if n != 0 { t.Errorf("origin received %d response.create; the turn must not be sent without its plugin", n) } + if n := fresh.Held(); n != 0 { + t.Errorf("the refused turn still holds the pipeline %d times", n) + } } // A turn over the 32 MiB a request body may be ends the socket with 1009, as HTTPS diff --git a/core/listener/forwardproxy/websocket_test.go b/core/listener/forwardproxy/websocket_test.go index c2209ae7c..52c6df186 100644 --- a/core/listener/forwardproxy/websocket_test.go +++ b/core/listener/forwardproxy/websocket_test.go @@ -1012,6 +1012,7 @@ func TestResponsesWebSocket_ReloadMidTurnFinishesOnTheTurnsPipeline(t *testing.T _, _ = fr.ReadFrame() }) c := h.dial(t, nil) + handshake := h.holder.Load().Held() c.send(responseCreate("hello")) <-midTurn @@ -1022,7 +1023,11 @@ func TestResponsesWebSocket_ReloadMidTurnFinishesOnTheTurnsPipeline(t *testing.T if err != nil { t.Fatal(err) } + old := h.holder.Load() h.holder.Store(fresh) + if n := old.Held(); n != handshake+1 { + t.Errorf("Held = %d mid-turn, want %d: the open turn must hold the pipeline it began on", n, handshake+1) + } close(resume) c.readUntil("response.completed") c.close() diff --git a/core/pipeline/context.go b/core/pipeline/context.go index dc80731b5..916043646 100644 --- a/core/pipeline/context.go +++ b/core/pipeline/context.go @@ -420,8 +420,9 @@ type Context struct { // stay open for minutes), and pctx.dispatched holds indexes into the // OLD pipeline's plugin slice — indexing the new one with them panics // when the new chain is shorter and otherwise runs hooks on plugins - // that never saw OnRequest. Direct *Pipeline calls (not via a Holder) - // never reload under a live request and leave this nil. + // that never saw OnRequest. Set by Holder.Run and Holder.Pin, which also + // hold the pipeline until RunFinish so a reload does not Stop it under + // the request. pinned *Pipeline // outcome is populated exactly once by Pipeline.RunFinish immediately diff --git a/core/pipeline/holder.go b/core/pipeline/holder.go index 67959b8ef..b6a4fadff 100644 --- a/core/pipeline/holder.go +++ b/core/pipeline/holder.go @@ -11,9 +11,8 @@ import ( // reloader Store()s a newly-built pipeline under the hood, and the // next Load returns it. atomic.Pointer guarantees a Load-then-use // sequence sees a fully-initialized pipeline — a request that started -// before a swap finishes against the original *Pipeline, and the -// reloader is responsible for not Stop()ing the old pipeline until a -// drain window expires. +// before a swap finishes against the original *Pipeline, which the +// reloader Stops once no request holds it (Pipeline.Drained). // // A Holder has exactly one long-lived value; its identity is the slot // listeners reference, not the pipeline at any given moment. @@ -40,13 +39,13 @@ func (h *Holder) Load() *Pipeline { return h.p.Load() } // new pipeline; in-flight requests that already Loaded the old one // continue to run against it. The caller owns the lifecycle of the // replaced pipeline — read the return value of the previous Load and -// call its Stop(ctx) after the intended drain window. +// call its Stop(ctx) once Drained is closed. func (h *Holder) Store(p *Pipeline) { h.p.Store(p) } -// Run is equivalent to h.Load().Run(ctx, pctx). Offered as a +// Run is equivalent to h.Pin(pctx).Run(ctx, pctx). Offered as a // one-liner so listeners can keep their Pipeline-era call sites // unchanged when they migrate their field type from *Pipeline to -// *Holder. One atomic Load per call. +// *Holder. // // Run also pins the loaded pipeline on pctx: every later phase on the // same request (RunResponse, RunResponseFrame, RunFinish, Resend) @@ -56,9 +55,20 @@ func (h *Holder) Store(p *Pipeline) { h.p.Store(p) } // the pin, a shorter new chain panics in RunFinish and an equal-length // one runs hooks on plugins that never saw the request. func (h *Holder) Run(ctx context.Context, pctx *Context) Action { - p := h.p.Load() - pctx.pinned = p - return p.Run(ctx, pctx) + return h.Pin(pctx).Run(ctx, pctx) +} + +// Pin pins pctx to the current pipeline and returns it. It re-checks the +// pipeline after holding it, so a reload that Stores in between is either +// seen here or sees the hold: the reloader counts holds only after its Store. +func (h *Holder) Pin(pctx *Context) *Pipeline { + for { + p := h.p.Load() + p.pin(pctx) + if h.p.Load() == p { + return p + } + } } // ForPctx returns the pipeline a pctx-carrying phase should run on: the diff --git a/core/pipeline/holder_reload_test.go b/core/pipeline/holder_reload_test.go index 685ae1be5..540330391 100644 --- a/core/pipeline/holder_reload_test.go +++ b/core/pipeline/holder_reload_test.go @@ -2,6 +2,7 @@ package pipeline import ( "context" + "sync" "testing" ) @@ -93,3 +94,105 @@ func TestHolder_CapabilityProbesFollowThePin(t *testing.T) { t.Error("HasStreamingResponders() = true, want false: the Holder itself serves new requests") } } + +// A request holds the pipeline it was pinned to once from Run until RunFinish, even +// when pinned again or finished twice, Drained closes on its release, and a +// finished request pinned again holds nothing. +func TestPipeline_HeldFromRunUntilRunFinish(t *testing.T) { + old := mustPipeline(t, &holderTestPlugin{name: "a"}) + h := NewHolder(old) + + pctx := &Context{Direction: Outbound} + h.Run(context.Background(), pctx) + h.Pin(pctx) + h.Store(mustPipeline(t, &holderTestPlugin{name: "b"})) + drained := old.Drained() + if got := old.Held(); got != 1 { + t.Fatalf("Held = %d during the request, want 1", got) + } + select { + case <-drained: + t.Fatal("Drained closed while a request held the pipeline") + default: + } + + h.RunFinish(context.Background(), pctx, Outcome{}) + h.RunFinish(context.Background(), pctx, Outcome{}) + if got := old.Held(); got != 0 { + t.Errorf("Held = %d after RunFinish, want 0", got) + } + select { + case <-drained: + default: + t.Error("Drained still open after the last request finished") + } + if p := h.Pin(pctx); p != h.Load() { + t.Error("a finished request pinned again stayed on the pipeline it finished on") + } + if got := h.Load().Held(); got != 0 { + t.Errorf("the new pipeline is held %d times by a request that had already finished", got) + } +} + +// heldDuringFinish records how many requests held its pipeline while it finished. +type heldDuringFinish struct { + holderTestPlugin + p *Pipeline + held int64 +} + +func (f *heldDuringFinish) OnFinish(context.Context, *Context) { f.held = f.p.Held() } + +// A request holds its pipeline until its Finishers return, so a reload cannot Stop +// a plugin that is still finishing. +func TestPipeline_FinishersRunWhileTheRequestHoldsThePipeline(t *testing.T) { + f := &heldDuringFinish{holderTestPlugin: holderTestPlugin{name: "f"}} + f.p = mustPipeline(t, f) + h := NewHolder(f.p) + pctx := &Context{Direction: Outbound} + h.Run(context.Background(), pctx) + h.RunFinish(context.Background(), pctx, Outcome{}) + if f.held != 1 { + t.Errorf("Held = %d inside OnFinish, want 1", f.held) + } +} + +// A request the reloader's wait-for-zero misses must not run on a stopped pipeline: +// Holder.Pin either holds the pipeline before the reloader counts it, or moves to +// the one Stored in its place. +func TestHolder_PinNeverLandsOnADrainedPipeline(t *testing.T) { + h := NewHolder(mustPipeline(t, &holderTestPlugin{name: "a"})) + var stopped sync.Map + done := make(chan struct{}) + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-done: + return + default: + } + pctx := &Context{Direction: Outbound} + p := h.Pin(pctx) + if _, ok := stopped.Load(p); ok { + t.Error("a request pinned a pipeline the reloader had already drained") + } + if p.Held() == 0 { + t.Error("Pin returned a pipeline the request does not hold") + } + h.RunFinish(context.Background(), pctx, Outcome{}) + } + }() + } + for range 500 { + old := h.Load() + h.Store(mustPipeline(t, &holderTestPlugin{name: "a"})) + <-old.Drained() + stopped.Store(old, true) + } + close(done) + wg.Wait() +} diff --git a/core/pipeline/pipeline.go b/core/pipeline/pipeline.go index 7e80eb2ef..db2aad2c3 100644 --- a/core/pipeline/pipeline.go +++ b/core/pipeline/pipeline.go @@ -4,6 +4,8 @@ import ( "context" "fmt" "log/slog" + "sync" + "sync/atomic" "time" ) @@ -32,6 +34,63 @@ type Pipeline struct { // requestWriters[i] is plugins[i]'s WritesRequestBody, read once at New for the // same reason. SetRequestModel is accepted only from a plugin that declares it. requestWriters []bool + // held counts requests pinned to this pipeline whose RunFinish has not run. + held atomic.Int64 + // drain is set by Drained and closed by the release that brings held to zero. + drain atomic.Pointer[drainSignal] +} + +type drainSignal struct { + once sync.Once + ch chan struct{} +} + +func (d *drainSignal) close() { d.once.Do(func() { close(d.ch) }) } + +// pin makes p the pipeline every later phase of pctx's request runs on, and holds p +// until that request's RunFinish. Listeners pin through Holder, which closes the race +// with a reload's Store. +func (p *Pipeline) pin(pctx *Context) { + if pctx.pinned == p { + return + } + if pctx.pinned != nil && !pctx.finished { + pctx.pinned.release() + } + pctx.pinned = p + if !pctx.finished { + p.held.Add(1) + } +} + +func (p *Pipeline) release() { + if p.held.Add(-1) == 0 { + if d := p.drain.Load(); d != nil { + p.closeIfIdle(d) + } + } +} + +// closeIfIdle closes d if nothing holds p now. The count is read again after d is +// seen: a release that hit zero before a request pinned p must not close it. +func (p *Pipeline) closeIfIdle(d *drainSignal) { + if p.held.Load() == 0 { + d.close() + } +} + +// Held is the number of requests pinned to p whose RunFinish has not run. +func (p *Pipeline) Held() int64 { return p.held.Load() } + +// Drained returns a channel closed once no request holds p. The reloader calls it +// after Storing p's replacement, when no new request can pin p. +func (p *Pipeline) Drained() <-chan struct{} { + d := &drainSignal{ch: make(chan struct{})} + if !p.drain.CompareAndSwap(nil, d) { + d = p.drain.Load() + } + p.closeIfIdle(d) + return d.ch } // Option configures pipeline construction. @@ -598,6 +657,10 @@ func (p *Pipeline) RunFinish(ctx context.Context, pctx *Context, outcome Outcome return } pctx.finished = true + if pctx.pinned == p { + // After the Finishers, which still run on p. + defer p.release() + } if len(pctx.dispatched) == 0 { return } diff --git a/core/reloader/reloader.go b/core/reloader/reloader.go index 994407bc7..94d3d2c0d 100644 --- a/core/reloader/reloader.go +++ b/core/reloader/reloader.go @@ -12,8 +12,9 @@ // changes to unreloadable fields like mode / listener addresses), // builds fresh pipelines via the caller-supplied PipelineBuilder, // runs their Start hooks under a bounded context, and Stores them -// into the Holders. The old pipelines keep running until a drain -// window expires, then Stop is invoked in a background goroutine. +// into the Holders. Each old pipeline keeps running until no request +// holds it, or the drain window expires, then Stop is invoked in a +// background goroutine. // // Validation failure at any stage leaves the current pipelines // untouched and records the error in Status so operators can see @@ -30,6 +31,7 @@ import ( "path/filepath" "reflect" "slices" + "sync" "sync/atomic" "time" @@ -53,8 +55,10 @@ type PipelineBuilder func() (inbound, outbound *pipeline.Pipeline, cfg *config.C // Option configures a Reloader at construction time. type Option func(*Reloader) -// WithDrainWindow overrides the delay before old pipelines are Stopped -// after a successful swap. Default 30s. Tests pass 0 to avoid sleeping. +// WithDrainWindow overrides the longest an old pipeline waits for the +// requests still pinned to it before it is Stopped anyway. Default 10m, +// so a tunnel left open for hours cannot keep its plugins alive. Tests +// pass 0 to stop at once. func WithDrainWindow(d time.Duration) Option { return func(r *Reloader) { r.drainWindow = d } } // WithDebounce overrides the debounce interval used to coalesce rapid @@ -142,7 +146,7 @@ func New(configPath string, inbound, outbound *pipeline.Holder, build PipelineBu inbound: inbound, outbound: outbound, build: build, - drainWindow: 30 * time.Second, + drainWindow: 10 * time.Minute, debounce: 250 * time.Millisecond, startTimeout: 60 * time.Second, } @@ -432,16 +436,21 @@ func (r *Reloader) reloadOnce(parent context.Context, watcher *fsnotify.Watcher) // deliberately: parent may cancel during shutdown, and we still // want a bounded Stop on the old plugins so they flush cleanly. go func() { - if r.drainWindow > 0 { - time.Sleep(r.drainWindow) - } - stopCtx, stopCancel := context.WithTimeout(context.Background(), 15*time.Second) - defer stopCancel() - if oldIn != nil { - oldIn.Stop(stopCtx) + deadline := time.Now().Add(r.drainWindow) + var remaining atomic.Int64 + var wg sync.WaitGroup + for _, p := range []*pipeline.Pipeline{oldIn, oldOut} { + if p != nil { + wg.Go(func() { remaining.Add(stopWhenUnheld(p, deadline)) }) + } } - if oldOut != nil { - oldOut.Stop(stopCtx) + wg.Wait() + if n := remaining.Load(); n > 0 { + // Routine while a client keeps a CONNECT tunnel or WebSocket open: the + // connection holds its pipeline until it closes. + slog.Warn("reloader: stopped old pipelines with requests still on them", + "sha256", hash[:12], "requests", n, "drainWindow", r.drainWindow) + return } slog.Info("reloader: drained old pipelines", "sha256", hash[:12]) }() @@ -452,6 +461,23 @@ func (r *Reloader) reloadOnce(parent context.Context, watcher *fsnotify.Watcher) r.recordSuccess(attempt, hash) } +// stopWhenUnheld Stops p once no request holds it, or at deadline, and returns how +// many requests still held it then. +func stopWhenUnheld(p *pipeline.Pipeline, deadline time.Time) int64 { + timer := time.NewTimer(time.Until(deadline)) + defer timer.Stop() + var n int64 + select { + case <-p.Drained(): + case <-timer.C: + n = p.Held() + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + p.Stop(ctx) + return n +} + // hash is what decides whether a reload has anything to do: the config's bytes, // and, with WithExtraWatch, the fingerprint of everything else the pipelines are // built from. While the fingerprint is empty it is the config's sha256, as it diff --git a/core/reloader/reloader_test.go b/core/reloader/reloader_test.go index 7e89d2a49..c017a71d9 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -6,8 +6,10 @@ import ( "crypto/sha256" "encoding/hex" "errors" + "log/slog" "os" "path/filepath" + "strings" "sync/atomic" "testing" "time" @@ -968,3 +970,136 @@ func TestValidateReloadable_RefusesMemoryEdit(t *testing.T) { t.Errorf("unchanged memory block was refused: %v", err) } } + +// shutdownPlugin records how many requests held its pipeline when it was stopped. +type shutdownPlugin struct { + p *pipeline.Pipeline + held int64 + stopped chan struct{} +} + +func (*shutdownPlugin) Name() string { return "shutdown" } +func (*shutdownPlugin) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{} +} +func (*shutdownPlugin) OnRequest(context.Context, *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} +func (*shutdownPlugin) OnResponse(context.Context, *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} +func (s *shutdownPlugin) Shutdown(context.Context) error { + s.held = s.p.Held() + close(s.stopped) + return nil +} + +// storeShutdownPipeline Stores a pipeline of one shutdownPlugin in h. +func storeShutdownPipeline(t *testing.T, h *pipeline.Holder) *shutdownPlugin { + t.Helper() + sp := &shutdownPlugin{stopped: make(chan struct{})} + p, err := pipeline.New([]pipeline.Plugin{sp}) + if err != nil { + t.Fatalf("pipeline.New: %v", err) + } + sp.p = p + h.Store(p) + return sp +} + +// drainLog passes on the reloader's closing record for the old pipelines of the +// reload whose config hashes to sha, ignoring other reloads still draining. +type drainLog struct { + sha string + records chan slog.Record +} + +func (drainLog) Enabled(context.Context, slog.Level) bool { return true } +func (d drainLog) Handle(_ context.Context, r slog.Record) error { + ours := false + r.Attrs(func(a slog.Attr) bool { + ours = ours || (a.Key == "sha256" && a.Value.String() == d.sha) + return true + }) + if ours && strings.Contains(r.Message, "old pipelines") { + d.records <- r.Clone() + } + return nil +} +func (d drainLog) WithAttrs([]slog.Attr) slog.Handler { return d } +func (d drainLog) WithGroup(string) slog.Handler { return d } + +// Each old pipeline is stopped once its own last request finishes, and the drain +// window stops it anyway under a request that outlasts it, warning how many it cut off. +func TestReloader_StopsAnOldPipelineWhenNoRequestHoldsIt(t *testing.T) { + for _, tc := range []struct { + name string + window time.Duration + finish bool + wantHeld int64 + wantLog string + }{ + {name: "the request's finish releases it", window: time.Hour, finish: true, wantHeld: 0, + wantLog: "reloader: drained old pipelines"}, + {name: "the window stops it under an open request", window: 0, wantHeld: 1, + wantLog: "reloader: stopped old pipelines with requests still on them"}, + } { + t.Run(tc.name, func(t *testing.T) { + content := "mode: envoy-sidecar\n# " + tc.name + "\n" + sum := sha256.Sum256([]byte(content)) + logs := drainLog{sha: hex.EncodeToString(sum[:])[:12], records: make(chan slog.Record, 1)} + prev := slog.Default() + slog.SetDefault(slog.New(logs)) + t.Cleanup(func() { slog.SetDefault(prev) }) + + committed := make(chan struct{}, 1) + _, b, cfgPath, inH, outH := setupWith(t, WithDrainWindow(tc.window), + WithOnCommit(func(*config.Config) { committed <- struct{}{} })) + idle, busy := storeShutdownPipeline(t, inH), storeShutdownPipeline(t, outH) + + pctx := &pipeline.Context{Direction: pipeline.Outbound} + outH.Run(context.Background(), pctx) + + b.set(builderResult{inbound: emptyPipeline(t), outbound: emptyPipeline(t), cfg: &config.Config{Mode: "envoy-sidecar"}}) + writeConfig(t, cfgPath, content) + <-committed + // The idle inbound pipeline stops while the outbound one is still held. + <-idle.stopped + if tc.finish { + outH.RunFinish(context.Background(), pctx, pipeline.Outcome{}) + // Checked before the wait below, so a hold RunFinish keeps fails here + // rather than leaving the reloader nothing to stop on. + select { + case <-busy.p.Drained(): + default: + t.Fatalf("the old pipeline is still held %d times after the request finished", busy.p.Held()) + } + } + <-busy.stopped + if busy.held != tc.wantHeld { + t.Errorf("stopped with %d requests holding it, want %d", busy.held, tc.wantHeld) + } + rec := <-logs.records + requests := int64(0) + rec.Attrs(func(a slog.Attr) bool { + if a.Key == "requests" { + requests = a.Value.Int64() + } + return true + }) + if rec.Message != tc.wantLog || requests != tc.wantHeld { + t.Errorf("logged %q with requests=%d, want %q with requests=%d", + rec.Message, requests, tc.wantLog, tc.wantHeld) + } + }) + } +} + +// Without WithDrainWindow, an old pipeline waits up to 10 minutes for its requests. +func TestNew_DrainWindowDefaultsToTenMinutes(t *testing.T) { + r := New("config.yaml", pipeline.NewHolder(emptyPipeline(t)), pipeline.NewHolder(emptyPipeline(t)), + (&fakeBuilder{}).build, &config.Config{Mode: "envoy-sidecar"}) + if r.drainWindow != 10*time.Minute { + t.Errorf("drainWindow = %v, want 10m", r.drainWindow) + } +} diff --git a/docs/framework-architecture.md b/docs/framework-architecture.md index 059d84d90..c0cd4eff1 100644 --- a/docs/framework-architecture.md +++ b/docs/framework-architecture.md @@ -853,7 +853,7 @@ curl http://localhost:9093/config # now-active config 5. Compare the new config to the active one on unreloadable fields (`Mode`, `Listener.*`, `Session.*`, `CostLedger.*`, `MTLS.*`); refuse if they differ (see below). 6. `Start` the new pipelines with a 60 s budget. On Start failure, `Stop` any partially-started pipelines so their goroutines don't leak. 7. `inboundH.Store(newIn)` + `outboundH.Store(newOut)` — new requests now route to the new pipelines. -8. Background goroutine: `time.Sleep(drainWindow)` (default 30 s) → `oldPipeline.Stop(ctx)` with a 15 s budget. In-flight requests that already Loaded the old pipeline finish against it. +8. Background goroutine: each old pipeline waits until no request holds it (`Pipeline.Drained`; a request holds from `Run` to `RunFinish`) or the drain window (default 10 min) expires → `oldPipeline.Stop(ctx)` with a 15 s budget. In-flight requests pinned to an old pipeline finish against it; a WARN names how many were still open when the window stopped it. A CONNECT tunnel or WebSocket holds its pipeline until the connection closes, so with a client that keeps one open, the full window and the WARN are the usual outcome, and rapid reloads keep several old pipelines' plugins running until their windows end. 9. `Status.LastSuccess`, `ReloadsOK`, `ActiveConfigSHA256` update atomically. **What's reloadable, what isn't** diff --git a/docs/lineage-wire-contract.md b/docs/lineage-wire-contract.md index 38953935d..cefc3c2cd 100644 --- a/docs/lineage-wire-contract.md +++ b/docs/lineage-wire-contract.md @@ -60,9 +60,9 @@ One HTTP exchange through the sidecar produces two OTLP spans. emitting the response span was recovered by the pipeline (a WARN is logged); the response span was emitted but lost — the two halves enter a batching exporter an exchange apart, so a response can be lost after its request has flushed; or the exchange outlived a config hot-reload — old - pipelines stop a drain window (default 30 s) after the swap, and a response span emitted into - the old, already-shut-down provider is dropped, which is routine for SSE or LLM exchanges - longer than the window. The consumer renders it as in-flight, never as a wrong + pipelines stop once no request holds them or the drain window (default 10 min) expires, and a + response span emitted into the old, already-shut-down provider is dropped, which happens only + for exchanges longer than the window. The consumer renders it as in-flight, never as a wrong pairing. A response span whose `lineage.outcome` is absent derives with `error` NULL (honest unknown), never `false`. - **Scope of `denied`.** The lineage plugin runs after the gate plugins and the pipeline