mirror of
https://github.com/zap-proto/zip.git
synced 2026-10-19 15:38:57 +00:00
fix(shutdown): bound the HTTP connection drain by the shutdown context
This commit is contained in:
@@ -56,7 +56,7 @@ Patch tags, one above the highest `v*` on GitHub. Push the commit to main, read
|
||||
- The call-plane body is ZAP, laid out from struct field order (`internal/zapenc`). Append fields at the end. Reordering, inserting or retyping a field changes the wire.
|
||||
- A pointer to a scalar, string or `[]byte` crosses as `list<T>` of at most one element: nil is the empty list, a pointer to zero is one zero element. A pointer to a struct is its message in a `bytes` slot, present when the slot is not null. `zapenc`, the `Layouts` emitter and the Rust projector (`TypeRef.Opt`) all follow this. A `*[N]byte` stays inline, so a pointer to an all-zero array reads back nil. Lists decode in one pass (`List.EachBytes`) and grow from the entries present, never the claimed count.
|
||||
- A plugin request holds its instance (`hold`, `instance.busy`) until the reply is written, a streamed body is closed (zap-proto/http closes it when the connection ends, v0.3.12+), or an upgrade's tunnel ends; an upgrade over ZAP is refused 501 because ZAP's server never runs a hijack, and a declined upgrade's reply is read as a reply, bounded by `declineWait`. `evict` raises `closing` before it reads `busy`; `claim` raises `busy` before it reads `closing`. Eviction never stops a busy instance and takes idle ones coldest first by when their last request ended, skipping one whose lock is held. A starter waits for room until its own deadline (`Plugin.Start` from when it arrived), then 503; the ceiling is exceeded only when no process may be stopped and no start is in flight. `lazy`/`idle`/`start` are copied off the spec at Load because the request path reads them unlocked.
|
||||
- Stopping a child is `terminate`: SIGTERM, then `Plugin.Grace` (default 20 s) for it to exit on its own, then SIGKILL. Every stop goes through it (Shutdown, Reload, Unload, eviction, a failed start). A plugin's shutdown hook only signals (`retire`); the outermost `Shutdown` waits for every child at once (`awaitChildren`), so N children cost the longest grace, not the sum. That wait ignores ctx: the drain usually spends it, and a host that returned first would be torn down under its children. A composed app's shutdown runs as its parent's hook with `nestedShutdown` on the context and leaves the wait to the parent.
|
||||
- Stopping a child is `terminate`: SIGTERM, then `Plugin.Grace` (default 20 s) for it to exit on its own, then SIGKILL. Every stop goes through it (Shutdown, Reload, Unload, eviction, a failed start). A plugin's shutdown hook only signals (`retire`); the outermost `Shutdown` waits for every child at once (`awaitChildren`), so N children cost the longest grace, not the sum. That wait ignores ctx: the drain usually spends it, and a host that returned first would be torn down under its children. Step 1 of `shutdown` stops every listener, then drains the HTTP transports' connections (`drainer`, fasthttp `ShutdownWithContext`) until ctx ends, so a stream a client keeps reading cannot hold a bounded shutdown. A composed app's shutdown runs as its parent's hook with `nestedShutdown` on the context and leaves the wait to the parent.
|
||||
- Doc prose is filed under `DocKey(pkg, method, path)`. zipdoc writes the package's import path. `pkgOf` reads the handler's package from its runtime name and maps `main` to the build info path, because a built binary names package main's functions `main.` while `go test` names them by import path.
|
||||
- MCP over HTTP: POST only (GET and DELETE answer 405), one message per request, no sessions, no batches, notifications 202. `negotiate` answers `initialize` in the requested revision when it is 2025-06-18, 2025-11-25 or 2026-07-28, else in 2026-07-28. The official TypeScript (1.30.0) and Python (2.2.0) clients refuse an answer in a revision they did not request and do not know.
|
||||
- `AdaptNetHTTP` carries an upgrade: a request with `Connection: Upgrade` runs the handler inside fasthttp's hijack, on a writer that is an `http.Hijacker`, so a net/http WebSocket server (coder/websocket, gorilla) writes its own 101 and owns the socket; a handler that declines answers an ordinary HTTP/1.1 reply. `wsx` stays the native way for a zip handler.
|
||||
|
||||
+8
-5
@@ -43,9 +43,11 @@ func (a *App) OnShutdown(fn func(context.Context) error) {
|
||||
// hookMu (snapshotting and clearing the hooks) and every later call returns
|
||||
// nil without repeating any step, so hooks run at most once.
|
||||
//
|
||||
// Sequence: (1) stop every transport accepting new connections; (2) drain
|
||||
// in-flight requests via fiber, bounded by ctx; (3) run teardown hooks LIFO,
|
||||
// passing ctx to each. All errors are joined and returned.
|
||||
// Sequence: (1) stop every transport accepting new connections, and wait for
|
||||
// the connections an HTTP transport holds, bounded by ctx; (2) drain in-flight
|
||||
// requests via fiber, bounded by ctx; (3) run teardown hooks LIFO, passing ctx
|
||||
// to each; (4) wait for every plugin child to exit. All errors are joined and
|
||||
// returned.
|
||||
func (a *App) shutdown(ctx context.Context) error {
|
||||
a.hookMu.Lock()
|
||||
if a.shuttingDown {
|
||||
@@ -57,8 +59,9 @@ func (a *App) shutdown(ctx context.Context) error {
|
||||
a.hooks = nil
|
||||
a.hookMu.Unlock()
|
||||
|
||||
// 1. Stop accepting new connections on every transport listener.
|
||||
a.closeServers()
|
||||
// 1. Stop accepting new connections on every transport listener, and drain
|
||||
// the connections HTTP holds until ctx ends.
|
||||
a.closeServers(ctx)
|
||||
a.closeMCP()
|
||||
// 2. Drain in-flight requests (graceful; bounded by ctx).
|
||||
// A program that was never served has no router to drain.
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
package zip_test
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
@@ -116,3 +119,49 @@ func TestShutdown_OutlivesAnEndedContext(t *testing.T) {
|
||||
t.Fatalf("the host returned before its child finished: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A stream a client keeps reading does not hold a bounded shutdown past its
|
||||
// context: the HTTP transport drains its connections only until ctx ends.
|
||||
func TestShutdownWithContext_DoesNotWaitOutAStream(t *testing.T) {
|
||||
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
|
||||
app.Raw(http.MethodGet, "/stream", func(c *zip.Ctx) error {
|
||||
return c.SendStreamWriter(func(w *bufio.Writer) {
|
||||
for {
|
||||
if _, err := w.WriteString(": ping\n\n"); err != nil || w.Flush() != nil {
|
||||
return
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
})
|
||||
})
|
||||
at := filepath.Join(t.TempDir(), "s.sock")
|
||||
go func() { _ = app.Listen("http://" + at) }()
|
||||
hc := &http.Client{Transport: &http.Transport{
|
||||
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
|
||||
return (&net.Dialer{}).DialContext(ctx, "unix", at)
|
||||
},
|
||||
}}
|
||||
var resp *http.Response
|
||||
for range 200 {
|
||||
if r, err := hc.Get("http://host/stream"); err == nil {
|
||||
resp = r
|
||||
break
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
if resp == nil {
|
||||
t.Fatal("stream never opened")
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if _, err := bufio.NewReader(resp.Body).ReadString('\n'); err != nil {
|
||||
t.Fatalf("read stream: %v", err)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
|
||||
defer cancel()
|
||||
start := time.Now()
|
||||
_ = app.ShutdownWithContext(ctx)
|
||||
if took := time.Since(start); took > 2*time.Second {
|
||||
t.Fatalf("ShutdownWithContext took %s with a 200ms context: an open stream held it", took)
|
||||
}
|
||||
}
|
||||
|
||||
+26
-1
@@ -670,7 +670,7 @@ func makeSocketDir(addr string) error {
|
||||
}
|
||||
|
||||
// closeServers stops every running listener. Called from Shutdown.
|
||||
func (a *App) closeServers() {
|
||||
func (a *App) closeServers(ctx context.Context) {
|
||||
a.srvMu.Lock()
|
||||
servers := a.servers
|
||||
unbind := a.unbind
|
||||
@@ -683,9 +683,28 @@ func (a *App) closeServers() {
|
||||
for _, u := range unbind {
|
||||
u()
|
||||
}
|
||||
// Every listener stops accepting first; then the servers that hold
|
||||
// connections open drain them together, bounded by ctx. A stream a client
|
||||
// keeps reading would otherwise hold the shutdown for as long as it liked.
|
||||
var drains []drainer
|
||||
for _, s := range servers {
|
||||
if d, ok := s.(drainer); ok {
|
||||
drains = append(drains, d)
|
||||
continue
|
||||
}
|
||||
_ = s.Close()
|
||||
}
|
||||
var wg sync.WaitGroup
|
||||
for _, d := range drains {
|
||||
wg.Go(func() { _ = d.ShutdownWithContext(ctx) })
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// drainer is a server that stops accepting and then waits, bounded by ctx, for
|
||||
// the connections it holds to finish.
|
||||
type drainer interface {
|
||||
ShutdownWithContext(ctx context.Context) error
|
||||
}
|
||||
|
||||
// splitScheme splits "scheme://addr" into (scheme, addr); a bare address
|
||||
@@ -726,6 +745,12 @@ func (h *httpServer) ListenAndServe() error {
|
||||
}
|
||||
func (h *httpServer) Close() error { return h.srv.Shutdown() }
|
||||
|
||||
// ShutdownWithContext stops accepting and waits for open connections, until ctx
|
||||
// ends.
|
||||
func (h *httpServer) ShutdownWithContext(ctx context.Context) error {
|
||||
return h.srv.ShutdownWithContext(ctx)
|
||||
}
|
||||
|
||||
// zapServer adapts zap-proto/http.Server the same way, and for the same
|
||||
// reason the HTTP one exists: the mode a socket is bound with is the App's to
|
||||
// decide and the upstream server takes an address, binds it, and never learns
|
||||
|
||||
Reference in New Issue
Block a user