From 5e003f121f2bb68780918b720e224515a65f319a Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Thu, 5 Nov 2026 09:19:20 -0700 Subject: [PATCH 1/5] :bug: Stop live and draining pipelines on shutdown Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- cmd/cortex-envoy/main.go | 3 +- cmd/cortex/main.go | 3 +- core/reloader/reloader.go | 50 ++++++++++++++++++++++++++++------ core/reloader/reloader_test.go | 29 ++++++++++++++++++++ docs/framework-architecture.md | 5 ++-- 5 files changed, 75 insertions(+), 15 deletions(-) diff --git a/cmd/cortex-envoy/main.go b/cmd/cortex-envoy/main.go index a47c84630..29078c005 100644 --- a/cmd/cortex-envoy/main.go +++ b/cmd/cortex-envoy/main.go @@ -355,8 +355,7 @@ func main() { sessionAPISrv.Shutdown(shutdownCtx) } - outboundPipeline.Stop(shutdownCtx) - inboundPipeline.Stop(shutdownCtx) + rld.Close(shutdownCtx) if sessions != nil { sessions.Close() diff --git a/cmd/cortex/main.go b/cmd/cortex/main.go index 866602861..13c701c12 100644 --- a/cmd/cortex/main.go +++ b/cmd/cortex/main.go @@ -1285,8 +1285,7 @@ func main() { _ = sessionAPISrv.Server().Close() } } - outboundPipeline.Stop(shutdownCtx) - inboundPipeline.Stop(shutdownCtx) + rld.Close(shutdownCtx) // Flushed AFTER both pipelines have stopped and before the store closes, so the // final minute includes every event a draining request still produced. Any earlier diff --git a/core/reloader/reloader.go b/core/reloader/reloader.go index 94d3d2c0d..ceded5e64 100644 --- a/core/reloader/reloader.go +++ b/core/reloader/reloader.go @@ -13,8 +13,8 @@ // builds fresh pipelines via the caller-supplied PipelineBuilder, // runs their Start hooks under a bounded context, and Stores them // 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. +// holds it, the drain window expires, or Close is called, then Stop is +// invoked in a background goroutine. Close also Stops the live pipelines. // // Validation failure at any stage leaves the current pipelines // untouched and records the error in Status so operators can see @@ -135,6 +135,15 @@ type Reloader struct { lastHash string // startData is the config file as New read it. startData []byte + + // cancelLoop and loopDone stop and await the watch goroutine; set by Start. + cancelLoop context.CancelFunc + loopDone chan struct{} + // Close sets closeCtx, then closes closing to stop every old pipeline still + // draining; drains counts their goroutines. + closing chan struct{} + closeCtx context.Context + drains sync.WaitGroup } // New constructs a Reloader. initialCfg must be the config the caller @@ -149,6 +158,7 @@ func New(configPath string, inbound, outbound *pipeline.Holder, build PipelineBu drainWindow: 10 * time.Minute, debounce: 250 * time.Millisecond, startTimeout: 60 * time.Second, + closing: make(chan struct{}), } for _, opt := range opts { opt(r) @@ -170,7 +180,7 @@ func New(configPath string, inbound, outbound *pipeline.Holder, build PipelineBu // Start arms the watcher. Blocks until the watcher is ready or fails // to arm (invalid directory, fsnotify init error). Returns nil on -// success; the watch goroutine then runs until ctx is cancelled. +// success; the watch goroutine then runs until ctx is cancelled or Close. // // Safe to call at most once. A Reloader whose Start returned an error // is not usable — construct a new one. @@ -240,7 +250,12 @@ func (r *Reloader) Start(ctx context.Context) error { r.seedHash() } - go r.watchLoop(ctx, watcher) + ctx, r.cancelLoop = context.WithCancel(ctx) + r.loopDone = make(chan struct{}) + go func() { + defer close(r.loopDone) + r.watchLoop(ctx, watcher) + }() slog.Info("reloader: watching for config changes", "path", r.configPath, "dir", dir, "drainWindow", r.drainWindow) return nil @@ -435,13 +450,15 @@ func (r *Reloader) reloadOnce(parent context.Context, watcher *fsnotify.Watcher) // Drain old pipelines in the background. Using context.Background // deliberately: parent may cancel during shutdown, and we still // want a bounded Stop on the old plugins so they flush cleanly. + r.drains.Add(1) go func() { + defer r.drains.Done() 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)) }) + wg.Go(func() { remaining.Add(r.stopWhenUnheld(p, deadline)) }) } } wg.Wait() @@ -461,9 +478,9 @@ 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 { +// stopWhenUnheld Stops p once no request holds it, at deadline, or on Close, and +// returns how many requests still held it then. +func (r *Reloader) stopWhenUnheld(p *pipeline.Pipeline, deadline time.Time) int64 { timer := time.NewTimer(time.Until(deadline)) defer timer.Stop() var n int64 @@ -471,6 +488,9 @@ func stopWhenUnheld(p *pipeline.Pipeline, deadline time.Time) int64 { case <-p.Drained(): case <-timer.C: n = p.Held() + case <-r.closing: + p.Stop(r.closeCtx) + return p.Held() } ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() @@ -478,6 +498,20 @@ func stopWhenUnheld(p *pipeline.Pipeline, deadline time.Time) int64 { return n } +// Close stops reloading, then Stops the live pipelines and every old one still +// draining, bounded by ctx. Call it once, after the listeners have stopped. +func (r *Reloader) Close(ctx context.Context) { + if r.cancelLoop != nil { + r.cancelLoop() + <-r.loopDone + } + r.closeCtx = ctx + close(r.closing) + r.outbound.Load().Stop(ctx) + r.inbound.Load().Stop(ctx) + r.drains.Wait() +} + // 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 c017a71d9..5d94633f3 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -1095,6 +1095,35 @@ func TestReloader_StopsAnOldPipelineWhenNoRequestHoldsIt(t *testing.T) { } } +// Close stops the live pipelines and an old one still draining under an open +// request, each once, before it returns. +func TestReloader_CloseStopsLiveAndDrainingPipelines(t *testing.T) { + committed := make(chan struct{}, 1) + r, b, cfgPath, _, outH := setupWith(t, WithDrainWindow(time.Hour), + WithOnCommit(func(*config.Config) { committed <- struct{}{} })) + draining := storeShutdownPipeline(t, outH) + outH.Run(context.Background(), &pipeline.Context{Direction: pipeline.Outbound}) + + liveIn, liveOut := pipeline.NewHolder(emptyPipeline(t)), pipeline.NewHolder(emptyPipeline(t)) + in, out := storeShutdownPipeline(t, liveIn), storeShutdownPipeline(t, liveOut) + b.set(builderResult{inbound: liveIn.Load(), outbound: liveOut.Load(), cfg: &config.Config{Mode: "envoy-sidecar"}}) + writeConfig(t, cfgPath, "mode: envoy-sidecar\n# reloaded\n") + <-committed + + // A second Stop of any of them would panic closing its stopped channel. + r.Close(context.Background()) + for name, sp := range map[string]*shutdownPlugin{"inbound": in, "outbound": out, "draining": draining} { + select { + case <-sp.stopped: + default: + t.Errorf("the %s pipeline was not stopped by Close", name) + } + } + if draining.held != 1 { + t.Errorf("the draining pipeline stopped with %d requests holding it, want 1", draining.held) + } +} + // 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)), diff --git a/docs/framework-architecture.md b/docs/framework-architecture.md index c0cd4eff1..84707abdb 100644 --- a/docs/framework-architecture.md +++ b/docs/framework-architecture.md @@ -460,8 +460,7 @@ if err := outboundPipeline.Start(initCtx); err != nil { // ... serve traffic ... // After listeners have drained (a pod) or closed (a local install) on SIGTERM: -outboundPipeline.Stop(shutdownCtx) // reverse order within each pipeline -inboundPipeline.Stop(shutdownCtx) +rld.Close(shutdownCtx) // the live pipelines, and any a reload is still draining ``` Semantics: @@ -853,7 +852,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: 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. +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. `Reloader.Close` at shutdown stops them at once. 9. `Status.LastSuccess`, `ReloadsOK`, `ActiveConfigSHA256` update atomically. **What's reloadable, what isn't** From ca0eff153c8cca468db9f56308c142ca57385780 Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Thu, 5 Nov 2026 09:27:32 -0700 Subject: [PATCH 2/5] :goal_net: Make sure Close doesn't run past shutdown deadline Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- cmd/cortex/main.go | 2 +- core/reloader/reloader.go | 16 +++++++++++---- core/reloader/reloader_test.go | 36 ++++++++++++++++++++++++++++------ 3 files changed, 43 insertions(+), 11 deletions(-) diff --git a/cmd/cortex/main.go b/cmd/cortex/main.go index 13c701c12..ff97e0721 100644 --- a/cmd/cortex/main.go +++ b/cmd/cortex/main.go @@ -1287,7 +1287,7 @@ func main() { } rld.Close(shutdownCtx) - // Flushed AFTER both pipelines have stopped and before the store closes, so the + // Flushed AFTER every pipeline has stopped and before the store closes, so the // final minute includes every event a draining request still produced. Any earlier // and a request finishing during the pipeline drain would land in a minute already // written; any later and the store is gone. diff --git a/core/reloader/reloader.go b/core/reloader/reloader.go index ceded5e64..c58cdca08 100644 --- a/core/reloader/reloader.go +++ b/core/reloader/reloader.go @@ -447,9 +447,8 @@ func (r *Reloader) reloadOnce(parent context.Context, watcher *fsnotify.Watcher) "sha256", hash[:12], "drainWindow", r.drainWindow) - // Drain old pipelines in the background. Using context.Background - // deliberately: parent may cancel during shutdown, and we still - // want a bounded Stop on the old plugins so they flush cleanly. + // Drain old pipelines in the background, on their own Stop budget rather than + // parent's; Close cuts the drain short at shutdown. r.drains.Add(1) go func() { defer r.drains.Done() @@ -509,7 +508,16 @@ func (r *Reloader) Close(ctx context.Context) { close(r.closing) r.outbound.Load().Stop(ctx) r.inbound.Load().Stop(ctx) - r.drains.Wait() + // Not past ctx: an old pipeline the window already stopped is on its own budget. + drained := make(chan struct{}) + go func() { + r.drains.Wait() + close(drained) + }() + select { + case <-drained: + case <-ctx.Done(): + } } // hash is what decides whether a reload has anything to do: the config's bytes, diff --git a/core/reloader/reloader_test.go b/core/reloader/reloader_test.go index 5d94633f3..2ed03a19f 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -971,11 +971,13 @@ func TestValidateReloadable_RefusesMemoryEdit(t *testing.T) { } } -// shutdownPlugin records how many requests held its pipeline when it was stopped. +// shutdownPlugin records how many requests held its pipeline when it was stopped, +// and with block set, stays in Shutdown until block is closed. type shutdownPlugin struct { p *pipeline.Pipeline held int64 stopped chan struct{} + block chan struct{} } func (*shutdownPlugin) Name() string { return "shutdown" } @@ -991,13 +993,16 @@ func (*shutdownPlugin) OnResponse(context.Context, *pipeline.Context) pipeline.A func (s *shutdownPlugin) Shutdown(context.Context) error { s.held = s.p.Held() close(s.stopped) + if s.block != nil { + <-s.block + } return nil } // storeShutdownPipeline Stores a pipeline of one shutdownPlugin in h. -func storeShutdownPipeline(t *testing.T, h *pipeline.Holder) *shutdownPlugin { +func storeShutdownPipeline(t *testing.T, h *pipeline.Holder, block chan struct{}) *shutdownPlugin { t.Helper() - sp := &shutdownPlugin{stopped: make(chan struct{})} + sp := &shutdownPlugin{stopped: make(chan struct{}), block: block} p, err := pipeline.New([]pipeline.Plugin{sp}) if err != nil { t.Fatalf("pipeline.New: %v", err) @@ -1055,7 +1060,7 @@ func TestReloader_StopsAnOldPipelineWhenNoRequestHoldsIt(t *testing.T) { 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) + idle, busy := storeShutdownPipeline(t, inH, nil), storeShutdownPipeline(t, outH, nil) pctx := &pipeline.Context{Direction: pipeline.Outbound} outH.Run(context.Background(), pctx) @@ -1101,11 +1106,11 @@ func TestReloader_CloseStopsLiveAndDrainingPipelines(t *testing.T) { committed := make(chan struct{}, 1) r, b, cfgPath, _, outH := setupWith(t, WithDrainWindow(time.Hour), WithOnCommit(func(*config.Config) { committed <- struct{}{} })) - draining := storeShutdownPipeline(t, outH) + draining := storeShutdownPipeline(t, outH, nil) outH.Run(context.Background(), &pipeline.Context{Direction: pipeline.Outbound}) liveIn, liveOut := pipeline.NewHolder(emptyPipeline(t)), pipeline.NewHolder(emptyPipeline(t)) - in, out := storeShutdownPipeline(t, liveIn), storeShutdownPipeline(t, liveOut) + in, out := storeShutdownPipeline(t, liveIn, nil), storeShutdownPipeline(t, liveOut, nil) b.set(builderResult{inbound: liveIn.Load(), outbound: liveOut.Load(), cfg: &config.Config{Mode: "envoy-sidecar"}}) writeConfig(t, cfgPath, "mode: envoy-sidecar\n# reloaded\n") <-committed @@ -1124,6 +1129,25 @@ func TestReloader_CloseStopsLiveAndDrainingPipelines(t *testing.T) { } } +// Close returns by its ctx even while an old pipeline the window stopped is still +// shutting down on its own budget. +func TestReloader_CloseDoesNotOutwaitItsContext(t *testing.T) { + committed := make(chan struct{}, 1) + r, b, cfgPath, _, outH := setupWith(t, WithOnCommit(func(*config.Config) { committed <- struct{}{} })) + block := make(chan struct{}) + t.Cleanup(func() { close(block) }) + slow := storeShutdownPipeline(t, outH, block) + + b.set(builderResult{inbound: emptyPipeline(t), outbound: emptyPipeline(t), cfg: &config.Config{Mode: "envoy-sidecar"}}) + writeConfig(t, cfgPath, "mode: envoy-sidecar\n# slow\n") + <-committed + <-slow.stopped + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + r.Close(ctx) +} + // 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)), From 6b284682e6db9c1e15859757c08594608fc9beda Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Thu, 5 Nov 2026 09:59:22 -0700 Subject: [PATCH 3/5] :art: Address review comments Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- core/reloader/reloader.go | 46 +++++++++++++++++++++------------- core/reloader/reloader_test.go | 38 ++++++++++++++++++---------- 2 files changed, 54 insertions(+), 30 deletions(-) diff --git a/core/reloader/reloader.go b/core/reloader/reloader.go index c58cdca08..c81aa6e90 100644 --- a/core/reloader/reloader.go +++ b/core/reloader/reloader.go @@ -139,8 +139,8 @@ type Reloader struct { // cancelLoop and loopDone stop and await the watch goroutine; set by Start. cancelLoop context.CancelFunc loopDone chan struct{} - // Close sets closeCtx, then closes closing to stop every old pipeline still - // draining; drains counts their goroutines. + // Close sets closeCtx, then closes closing, to stop every old pipeline still + // draining on closeCtx; drains counts their goroutines. closing chan struct{} closeCtx context.Context drains sync.WaitGroup @@ -420,7 +420,7 @@ func (r *Reloader) reloadOnce(parent context.Context, watcher *fsnotify.Watcher) } if err := newOut.Start(startCtx); err != nil { // Inbound started cleanly; unwind it before bailing. - unwindCtx, unwindCancel := context.WithTimeout(context.Background(), 15*time.Second) + unwindCtx, unwindCancel := r.stopCtx() defer unwindCancel() newIn.Stop(unwindCtx) r.recordFailure(attempt, fmt.Errorf("outbound Start: %w", err)) @@ -488,34 +488,46 @@ func (r *Reloader) stopWhenUnheld(p *pipeline.Pipeline, deadline time.Time) int6 case <-timer.C: n = p.Held() case <-r.closing: - p.Stop(r.closeCtx) - return p.Held() + n = p.Held() } - ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + ctx, cancel := r.stopCtx() defer cancel() p.Stop(ctx) return n } +// stopCtx is the budget for a Stop: Close's ctx once Close has begun, else 15s. +func (r *Reloader) stopCtx() (context.Context, context.CancelFunc) { + select { + case <-r.closing: + return r.closeCtx, func() {} + default: + return context.WithTimeout(context.Background(), 15*time.Second) + } +} + // Close stops reloading, then Stops the live pipelines and every old one still -// draining, bounded by ctx. Call it once, after the listeners have stopped. +// draining. It returns when they are stopped or ctx is done, whichever is first; +// a plugin still shutting down then finishes in the background. Call it once, +// after the listeners have stopped. func (r *Reloader) Close(ctx context.Context) { - if r.cancelLoop != nil { - r.cancelLoop() - <-r.loopDone - } r.closeCtx = ctx close(r.closing) - r.outbound.Load().Stop(ctx) - r.inbound.Load().Stop(ctx) - // Not past ctx: an old pipeline the window already stopped is on its own budget. - drained := make(chan struct{}) + done := make(chan struct{}) go func() { + defer close(done) + // The watch loop first, so a reload in flight has committed or given up + // before the live pipelines are read. + if r.cancelLoop != nil { + r.cancelLoop() + <-r.loopDone + } + r.outbound.Load().Stop(ctx) + r.inbound.Load().Stop(ctx) r.drains.Wait() - close(drained) }() select { - case <-drained: + case <-done: case <-ctx.Done(): } } diff --git a/core/reloader/reloader_test.go b/core/reloader/reloader_test.go index 2ed03a19f..bca075321 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -971,11 +971,13 @@ func TestValidateReloadable_RefusesMemoryEdit(t *testing.T) { } } -// shutdownPlugin records how many requests held its pipeline when it was stopped, -// and with block set, stays in Shutdown until block is closed. +// shutdownPlugin records how many requests held its pipeline when it was stopped +// and whether its ctx was already done, and with block set, stays in Shutdown +// until block is closed. type shutdownPlugin struct { p *pipeline.Pipeline held int64 + ctxErr error stopped chan struct{} block chan struct{} } @@ -990,8 +992,8 @@ func (*shutdownPlugin) OnRequest(context.Context, *pipeline.Context) pipeline.Ac 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() +func (s *shutdownPlugin) Shutdown(ctx context.Context) error { + s.held, s.ctxErr = s.p.Held(), ctx.Err() close(s.stopped) if s.block != nil { <-s.block @@ -1129,23 +1131,33 @@ func TestReloader_CloseStopsLiveAndDrainingPipelines(t *testing.T) { } } -// Close returns by its ctx even while an old pipeline the window stopped is still -// shutting down on its own budget. +// Close returns by its ctx even while a plugin it is stopping is still shutting down. func TestReloader_CloseDoesNotOutwaitItsContext(t *testing.T) { - committed := make(chan struct{}, 1) - r, b, cfgPath, _, outH := setupWith(t, WithOnCommit(func(*config.Config) { committed <- struct{}{} })) + r, _, _, _, outH := setup(t) block := make(chan struct{}) t.Cleanup(func() { close(block) }) - slow := storeShutdownPipeline(t, outH, block) + storeShutdownPipeline(t, outH, block) - b.set(builderResult{inbound: emptyPipeline(t), outbound: emptyPipeline(t), cfg: &config.Config{Mode: "envoy-sidecar"}}) - writeConfig(t, cfgPath, "mode: envoy-sidecar\n# slow\n") - <-committed - <-slow.stopped + ctx, cancel := context.WithCancel(context.Background()) + cancel() + r.Close(ctx) +} +// Once Close has begun, an old pipeline is stopped on Close's ctx, even when its +// drain is what woke it. +func TestReloader_StopsOnTheCloseContextOnceClosing(t *testing.T) { + r, _, _, _, _ := setup(t) ctx, cancel := context.WithCancel(context.Background()) cancel() r.Close(ctx) + // Repeated because select picks among the ready drain and close cases at random. + for range 20 { + sp := storeShutdownPipeline(t, pipeline.NewHolder(emptyPipeline(t)), nil) + r.stopWhenUnheld(sp.p, time.Now().Add(time.Hour)) + if sp.ctxErr == nil { + t.Fatal("stopped on a fresh budget after Close began") + } + } } // Without WithDrainWindow, an old pipeline waits up to 10 minutes for its requests. From 2f851dd33ecc09513f1a202c335935418a28ad8a Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Thu, 5 Nov 2026 10:34:44 -0700 Subject: [PATCH 4/5] :goal_net: Read live pipelines Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- cmd/cortex/main.go | 5 +++-- core/reloader/reloader.go | 40 ++++++++++++++++++++++++---------- core/reloader/reloader_test.go | 36 +++++++++++++++++++++++++++++- 3 files changed, 66 insertions(+), 15 deletions(-) diff --git a/cmd/cortex/main.go b/cmd/cortex/main.go index ff97e0721..bc9702168 100644 --- a/cmd/cortex/main.go +++ b/cmd/cortex/main.go @@ -1287,8 +1287,9 @@ func main() { } rld.Close(shutdownCtx) - // Flushed AFTER every pipeline has stopped and before the store closes, so the - // final minute includes every event a draining request still produced. Any earlier + // Flushed AFTER rld.Close (every pipeline stopped, or the shutdown budget spent) + // and before the store closes, so the final minute includes every event a + // draining request still produced. Any earlier // and a request finishing during the pipeline drain would land in a minute already // written; any later and the store is gone. // diff --git a/core/reloader/reloader.go b/core/reloader/reloader.go index c81aa6e90..e5efe18b8 100644 --- a/core/reloader/reloader.go +++ b/core/reloader/reloader.go @@ -140,10 +140,12 @@ type Reloader struct { cancelLoop context.CancelFunc loopDone chan struct{} // Close sets closeCtx, then closes closing, to stop every old pipeline still - // draining on closeCtx; drains counts their goroutines. + // draining on closeCtx; drains counts their goroutines. commitMu orders a + // reload's commit against Close. closing chan struct{} closeCtx context.Context drains sync.WaitGroup + commitMu sync.Mutex } // New constructs a Reloader. initialCfg must be the config the caller @@ -430,12 +432,23 @@ func (r *Reloader) reloadOnce(parent context.Context, watcher *fsnotify.Watcher) // Commit: swap Holders and update active-config snapshot. From // here on, requests drain onto the new pipelines; old in-flight // requests keep their existing *Pipeline reference. + r.commitMu.Lock() + select { + case <-r.closing: + r.commitMu.Unlock() + newOut.Stop(r.closeCtx) + newIn.Stop(r.closeCtx) + return + default: + } oldIn := r.inbound.Load() oldOut := r.outbound.Load() r.inbound.Store(newIn) r.outbound.Store(newOut) r.activeCfg.Store(newCfg) r.lastHash = hash + r.drains.Add(1) + r.commitMu.Unlock() // After the swap, so a hook cannot observe a config that was refused, and before the log line, so // "pipelines swapped" means everything this reload applies has been applied. @@ -449,7 +462,6 @@ func (r *Reloader) reloadOnce(parent context.Context, watcher *fsnotify.Watcher) // Drain old pipelines in the background, on their own Stop budget rather than // parent's; Close cuts the drain short at shutdown. - r.drains.Add(1) go func() { defer r.drains.Done() deadline := time.Now().Add(r.drainWindow) @@ -507,24 +519,28 @@ func (r *Reloader) stopCtx() (context.Context, context.CancelFunc) { } // Close stops reloading, then Stops the live pipelines and every old one still -// draining. It returns when they are stopped or ctx is done, whichever is first; -// a plugin still shutting down then finishes in the background. Call it once, -// after the listeners have stopped. +// draining; a reload still building Stops its pipelines instead. It returns when +// they are stopped or ctx is done, whichever is first; a plugin still shutting +// down then finishes in the background. Call it once, after the listeners have +// stopped. func (r *Reloader) Close(ctx context.Context) { r.closeCtx = ctx + r.commitMu.Lock() close(r.closing) + liveIn, liveOut := r.inbound.Load(), r.outbound.Load() + r.commitMu.Unlock() + if r.cancelLoop != nil { + r.cancelLoop() + } done := make(chan struct{}) go func() { defer close(done) - // The watch loop first, so a reload in flight has committed or given up - // before the live pipelines are read. - if r.cancelLoop != nil { - r.cancelLoop() + liveOut.Stop(ctx) + liveIn.Stop(ctx) + r.drains.Wait() + if r.loopDone != nil { <-r.loopDone } - r.outbound.Load().Stop(ctx) - r.inbound.Load().Stop(ctx) - r.drains.Wait() }() select { case <-done: diff --git a/core/reloader/reloader_test.go b/core/reloader/reloader_test.go index bca075321..d091ea468 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -31,8 +31,12 @@ type fakeBuilder struct { // `next` at call time. next atomic.Value // holds builderResult cfg atomic.Value // holds *config.Config + // gate, when set, makes build signal entered and wait for release. + gate atomic.Pointer[buildGate] } +type buildGate struct{ entered, release chan struct{} } + type builderResult struct { inbound *pipeline.Pipeline outbound *pipeline.Pipeline @@ -48,6 +52,10 @@ func (b *fakeBuilder) set(r builderResult) { } func (b *fakeBuilder) build() (*pipeline.Pipeline, *pipeline.Pipeline, *config.Config, error) { + if g := b.gate.Load(); g != nil { + close(g.entered) + <-g.release + } r := b.next.Load().(builderResult) return r.inbound, r.outbound, r.cfg, r.err } @@ -1131,7 +1139,7 @@ func TestReloader_CloseStopsLiveAndDrainingPipelines(t *testing.T) { } } -// Close returns by its ctx even while a plugin it is stopping is still shutting down. +// Close returns by its ctx even while a live pipeline's plugin is still shutting down. func TestReloader_CloseDoesNotOutwaitItsContext(t *testing.T) { r, _, _, _, outH := setup(t) block := make(chan struct{}) @@ -1143,6 +1151,32 @@ func TestReloader_CloseDoesNotOutwaitItsContext(t *testing.T) { r.Close(ctx) } +// Close stops the live pipelines without waiting for a reload still building, and +// that reload stops its own pipelines instead of committing them. +func TestReloader_CloseDuringABuildStopsLiveAndBuiltPipelines(t *testing.T) { + r, b, cfgPath, inH, outH := setup(t) + live := storeShutdownPipeline(t, outH, nil) + g := &buildGate{entered: make(chan struct{}), release: make(chan struct{})} + b.gate.Store(g) + builtIn := storeShutdownPipeline(t, pipeline.NewHolder(emptyPipeline(t)), nil) + builtOut := storeShutdownPipeline(t, pipeline.NewHolder(emptyPipeline(t)), nil) + b.set(builderResult{inbound: builtIn.p, outbound: builtOut.p, cfg: &config.Config{Mode: "envoy-sidecar"}}) + writeConfig(t, cfgPath, "mode: envoy-sidecar\n# building\n") + <-g.entered + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + r.Close(ctx) + <-live.stopped // while the build is still blocked + + close(g.release) + <-builtIn.stopped + <-builtOut.stopped + if inH.Load() == builtIn.p || outH.Load() == builtOut.p { + t.Error("a reload committed its pipelines after Close began") + } +} + // Once Close has begun, an old pipeline is stopped on Close's ctx, even when its // drain is what woke it. func TestReloader_StopsOnTheCloseContextOnceClosing(t *testing.T) { From 762eef395ceb41722ee5b03b23d3a183e7152cc2 Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Thu, 5 Nov 2026 11:59:49 -0700 Subject: [PATCH 5/5] :goal_net: Stop live pipelines concurrently Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- core/reloader/reloader.go | 6 ++++-- core/reloader/reloader_test.go | 7 +++++-- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/core/reloader/reloader.go b/core/reloader/reloader.go index e5efe18b8..3d83d876b 100644 --- a/core/reloader/reloader.go +++ b/core/reloader/reloader.go @@ -535,8 +535,10 @@ func (r *Reloader) Close(ctx context.Context) { done := make(chan struct{}) go func() { defer close(done) - liveOut.Stop(ctx) - liveIn.Stop(ctx) + var live sync.WaitGroup + live.Go(func() { liveOut.Stop(ctx) }) + live.Go(func() { liveIn.Stop(ctx) }) + live.Wait() r.drains.Wait() if r.loopDone != nil { <-r.loopDone diff --git a/core/reloader/reloader_test.go b/core/reloader/reloader_test.go index d091ea468..28b1985ce 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -1139,16 +1139,19 @@ func TestReloader_CloseStopsLiveAndDrainingPipelines(t *testing.T) { } } -// Close returns by its ctx even while a live pipeline's plugin is still shutting down. +// Close returns by its ctx even while a live pipeline's plugin is still shutting +// down, and that plugin does not keep the other live pipeline from stopping. func TestReloader_CloseDoesNotOutwaitItsContext(t *testing.T) { - r, _, _, _, outH := setup(t) + r, _, _, inH, outH := setup(t) block := make(chan struct{}) t.Cleanup(func() { close(block) }) storeShutdownPipeline(t, outH, block) + in := storeShutdownPipeline(t, inH, nil) ctx, cancel := context.WithCancel(context.Background()) cancel() r.Close(ctx) + <-in.stopped } // Close stops the live pipelines without waiting for a reload still building, and