mirror of
https://github.com/zap-proto/zip.git
synced 2026-10-19 15:38:57 +00:00
feat(load): stop a child with SIGTERM and a 20s grace, all children at once
This commit is contained in:
@@ -56,6 +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. 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.
|
||||
|
||||
@@ -10,6 +10,8 @@ import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/zap-proto/zip"
|
||||
@@ -128,6 +130,17 @@ func main() {
|
||||
return
|
||||
}
|
||||
|
||||
// TESTPLUGIN_ON_TERM names a file the plugin writes when SIGTERM arrives,
|
||||
// TESTPLUGIN_TERM_DELAY after it, then exits — standing in for the stores a
|
||||
// real child closes and ships on the way out. TESTPLUGIN_TERM=ignore makes
|
||||
// it ignore SIGTERM, so only a kill ends it.
|
||||
switch {
|
||||
case os.Getenv("TESTPLUGIN_TERM") == "ignore":
|
||||
signal.Ignore(syscall.SIGTERM)
|
||||
case os.Getenv("TESTPLUGIN_ON_TERM") != "":
|
||||
go onTerm(os.Getenv("TESTPLUGIN_ON_TERM"))
|
||||
}
|
||||
|
||||
// Addr is the whole plugin side of the contract: serve where the host said.
|
||||
// Report rather than swallow: a fixture that discards this error exits
|
||||
// silently and the host reports only "exited before listening".
|
||||
@@ -135,3 +148,16 @@ func main() {
|
||||
log.Fatalf("testplugin: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// onTerm waits for SIGTERM, takes TESTPLUGIN_TERM_DELAY, writes path and exits.
|
||||
func onTerm(path string) {
|
||||
sig := make(chan os.Signal, 1)
|
||||
signal.Notify(sig, syscall.SIGTERM)
|
||||
<-sig
|
||||
d, _ := time.ParseDuration(os.Getenv("TESTPLUGIN_TERM_DELAY"))
|
||||
time.Sleep(d)
|
||||
if err := os.WriteFile(path, []byte("drained"), 0o600); err != nil {
|
||||
os.Exit(1)
|
||||
}
|
||||
os.Exit(0)
|
||||
}
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -69,8 +70,9 @@ import (
|
||||
// Re-registering on reload would grow the route table without bound.
|
||||
// - The new process must be listening before any request moves to it; the
|
||||
// old one keeps serving until then, and a failed start changes nothing.
|
||||
// - The old process is killed, reaped exactly once, its pooled connections
|
||||
// closed, and its directory removed — so nothing survives the swap.
|
||||
// - The old process is stopped (SIGTERM, then a kill once its grace is
|
||||
// spent), reaped exactly once, its pooled connections closed, and its
|
||||
// directory removed — so nothing survives the swap.
|
||||
|
||||
// AddrEnv is the variable a host sets to tell a plugin where to listen. A
|
||||
// plugin reads it through [Addr].
|
||||
@@ -127,6 +129,11 @@ type Plugin struct {
|
||||
// requests already in flight on it finish. Zero means 5s.
|
||||
Drain time.Duration
|
||||
|
||||
// Grace is how long a stopped process has to exit on its own after SIGTERM
|
||||
// before it is killed: the time it has to finish what it is serving, close
|
||||
// and ship its stores, and hand back what it holds. Zero means 20s.
|
||||
Grace time.Duration
|
||||
|
||||
// IdleAfter stops a LAZY plugin that has not served for this long, to be
|
||||
// started again by the next request that needs it. Zero means never, which
|
||||
// is the historical behaviour. See idle.go: without it a host pays for
|
||||
@@ -283,6 +290,7 @@ type instance struct {
|
||||
sock string
|
||||
client Client
|
||||
started time.Time
|
||||
grace time.Duration // Plugin.Grace, fixed when the process starts
|
||||
|
||||
// done is CLOSED when the child has exited, and exitErr holds why. A
|
||||
// closed channel broadcasts: the supervisor, a drain, and Shutdown can all
|
||||
@@ -339,6 +347,14 @@ func (p Plugin) startLimit() time.Duration {
|
||||
return p.Start
|
||||
}
|
||||
|
||||
// graceLimit is Grace with its default applied.
|
||||
func (p Plugin) graceLimit() time.Duration {
|
||||
if p.Grace <= 0 {
|
||||
return 20 * time.Second
|
||||
}
|
||||
return p.Grace
|
||||
}
|
||||
|
||||
// startError reports why the last start attempt failed, or "" when none has. It is
|
||||
// what separates "lazy and never asked" from "asked, and it will not come up" —
|
||||
// two states that are otherwise the same Running=false.
|
||||
@@ -605,16 +621,15 @@ func (a *App) load(prefixes []string, spec Plugin) error {
|
||||
go a.supervise(p, in)
|
||||
}
|
||||
|
||||
// The hook SIGNALS and returns; shutdown's last step waits for every
|
||||
// child at once. Each child drains on its own clock, so N children take the
|
||||
// longest grace rather than the sum of them, and the host still outlives
|
||||
// every one.
|
||||
a.OnShutdown(func(context.Context) error {
|
||||
p.mu.Lock()
|
||||
p.closed = true
|
||||
p.mu.Unlock()
|
||||
if cur := p.cur.Swap(nil); cur != nil {
|
||||
stop(cur, 0)
|
||||
}
|
||||
// Wait out any instance a Reload or Unload is still draining, so no
|
||||
// child survives this host.
|
||||
p.stopping.Wait()
|
||||
p.retire(p.cur.Swap(nil), 0)
|
||||
return nil
|
||||
})
|
||||
return nil
|
||||
@@ -784,7 +799,7 @@ func start(spec Plugin) (*instance, error) {
|
||||
|
||||
// Exactly one Wait for this process's lifetime; everyone else observes the
|
||||
// close. Two Waits is an error, and a child nobody waits on is a zombie.
|
||||
in := &instance{cmd: cmd, dir: dir, sock: sock, started: time.Now(), done: make(chan struct{})}
|
||||
in := &instance{cmd: cmd, dir: dir, sock: sock, started: time.Now(), grace: spec.graceLimit(), done: make(chan struct{})}
|
||||
go func() {
|
||||
in.exitErr = cmd.Wait()
|
||||
// Wait returns once the child's output is copied, so what is left is a
|
||||
@@ -873,24 +888,20 @@ func fetch(spec Plugin) (string, error) {
|
||||
// Both built-in clients implement it; a custom one need not.
|
||||
type idleCloser interface{ CloseIdleConnections() }
|
||||
|
||||
// stop releases everything an instance holds, after letting it serve for grace.
|
||||
// stop releases everything an instance holds, after letting it serve for drain.
|
||||
// Safe to call on a partially-started instance.
|
||||
func stop(in *instance, grace time.Duration) {
|
||||
func stop(in *instance, drain time.Duration) {
|
||||
if in == nil {
|
||||
return
|
||||
}
|
||||
if grace > 0 {
|
||||
if drain > 0 {
|
||||
select {
|
||||
case <-in.done: // already gone; nothing to wait out
|
||||
case <-time.After(grace):
|
||||
case <-time.After(drain):
|
||||
}
|
||||
}
|
||||
if in.cmd != nil && in.cmd.Process != nil {
|
||||
_ = in.cmd.Process.Kill()
|
||||
select {
|
||||
case <-in.done:
|
||||
case <-time.After(5 * time.Second):
|
||||
}
|
||||
terminate(in)
|
||||
}
|
||||
// Pooled connections to a dead socket are not reusable and would otherwise
|
||||
// be held by the transport until GC.
|
||||
@@ -902,6 +913,35 @@ func stop(in *instance, grace time.Duration) {
|
||||
}
|
||||
}
|
||||
|
||||
// terminate ends the process: SIGTERM, then up to its grace for it to exit,
|
||||
// then SIGKILL. A child that is asked runs its own shutdown — it drains, closes
|
||||
// its stores with a checkpoint and ships them — which a kill skips, leaving
|
||||
// hot -wal sidecars and unshipped writes behind. Where the platform has no
|
||||
// SIGTERM the signal fails and the kill is immediate, as before.
|
||||
func terminate(in *instance) {
|
||||
select {
|
||||
case <-in.done:
|
||||
return
|
||||
default:
|
||||
}
|
||||
grace := in.grace
|
||||
if grace <= 0 {
|
||||
grace = Plugin{}.graceLimit()
|
||||
}
|
||||
if in.cmd.Process.Signal(syscall.SIGTERM) == nil {
|
||||
select {
|
||||
case <-in.done:
|
||||
return
|
||||
case <-time.After(grace):
|
||||
}
|
||||
}
|
||||
_ = in.cmd.Process.Kill()
|
||||
select {
|
||||
case <-in.done:
|
||||
case <-time.After(5 * time.Second):
|
||||
}
|
||||
}
|
||||
|
||||
// waitListening blocks until the socket accepts, the child exits, or the
|
||||
// deadline passes. Watching the child matters: a plugin that dies immediately
|
||||
// should say so, not time out.
|
||||
|
||||
+43
-1
@@ -68,10 +68,52 @@ func (a *App) shutdown(ctx context.Context) error {
|
||||
}
|
||||
// 3. Tear subsystems down in reverse mount order (LIFO). Run ALL hooks
|
||||
// even if some fail; aggregate every error.
|
||||
nested := ctx.Value(nestedShutdown{}) != nil
|
||||
hookCtx := context.WithValue(ctx, nestedShutdown{}, true)
|
||||
for i := len(hooks) - 1; i >= 0; i-- {
|
||||
if err := hooks[i](ctx); err != nil {
|
||||
if err := hooks[i](hookCtx); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
// 4. Wait for every plugin child to exit. The hooks only signalled them, so
|
||||
// they drain together and each is killed at its own grace; the host
|
||||
// returns once the last is gone, or when ctx ends. A composed app's
|
||||
// shutdown runs as its parent's hook and leaves the wait to the
|
||||
// outermost one, which walks the whole composition: waiting at each level
|
||||
// would drain the children one after another.
|
||||
if !nested {
|
||||
if g, err := a.liveOrBuild(); err == nil {
|
||||
errs = append(errs, awaitChildren(ctx, g.hosts))
|
||||
}
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
// nestedShutdown marks the context a parent hands its teardown hooks, so a
|
||||
// composed app's shutdown knows it is not the outermost.
|
||||
type nestedShutdown struct{}
|
||||
|
||||
// awaitChildren blocks until no plugin in hosts has a process left to stop.
|
||||
func awaitChildren(ctx context.Context, hosts []*App) error {
|
||||
var ps []*plugin
|
||||
for _, h := range hosts {
|
||||
h.plugMu.Lock()
|
||||
for _, p := range h.plugins {
|
||||
ps = append(ps, p)
|
||||
}
|
||||
h.plugMu.Unlock()
|
||||
}
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
for _, p := range ps {
|
||||
p.stopping.Wait()
|
||||
}
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
package zip_test
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/zap-proto/zip"
|
||||
)
|
||||
|
||||
func alive(pid int) bool { return syscall.Kill(pid, 0) == nil }
|
||||
|
||||
// Shutdown asks each child to exit and waits for it, so a child that takes time
|
||||
// on the way out — closing and shipping its stores — gets to finish. A kill
|
||||
// would end it before it wrote the file.
|
||||
func TestShutdown_TermsChildBeforeKilling(t *testing.T) {
|
||||
bin := buildPlugin(t, "v1")
|
||||
mark := filepath.Join(t.TempDir(), "drained")
|
||||
|
||||
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
|
||||
app.Use(must(zip.Load(zip.Plugin{Name: "demo", Bin: bin, Env: []string{
|
||||
"TESTPLUGIN_ON_TERM=" + mark, "TESTPLUGIN_TERM_DELAY=300ms",
|
||||
}}, "/v1/demo")))
|
||||
version(t, app)
|
||||
|
||||
if err := app.Shutdown(); err != nil {
|
||||
t.Fatalf("Shutdown: %v", err)
|
||||
}
|
||||
b, err := os.ReadFile(mark)
|
||||
if err != nil || string(b) != "drained" {
|
||||
t.Fatalf("child did not finish its shutdown before the host returned: %q, %v", b, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A child that ignores SIGTERM is killed once its grace is spent, and the host
|
||||
// does not wait past it.
|
||||
func TestShutdown_KillsChildAfterGrace(t *testing.T) {
|
||||
bin := buildPlugin(t, "v1")
|
||||
|
||||
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
|
||||
app.Use(must(zip.Load(zip.Plugin{Name: "demo", Bin: bin, Grace: 300 * time.Millisecond,
|
||||
Env: []string{"TESTPLUGIN_TERM=ignore"}}, "/v1/demo")))
|
||||
version(t, app)
|
||||
pid := pidOf(app, "demo")
|
||||
if pid == 0 {
|
||||
t.Fatal("demo is not running")
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
if err := app.Shutdown(); err != nil {
|
||||
t.Fatalf("Shutdown: %v", err)
|
||||
}
|
||||
took := time.Since(start)
|
||||
if took < 300*time.Millisecond || took > 5*time.Second {
|
||||
t.Fatalf("Shutdown took %s, want the 300ms grace and then a kill", took)
|
||||
}
|
||||
if alive(pid) {
|
||||
t.Fatalf("child %d survived its host", pid)
|
||||
}
|
||||
}
|
||||
|
||||
// Children drain together: three that each take 600ms on the way out cost the
|
||||
// host one grace, not three.
|
||||
func TestShutdown_ChildrenDrainTogether(t *testing.T) {
|
||||
bin := buildPlugin(t, "v1")
|
||||
dir := t.TempDir()
|
||||
|
||||
app := zip.New(zip.Config{AppName: "host", DisableStartupMessage: true})
|
||||
for i := range 3 {
|
||||
n := strconv.Itoa(i)
|
||||
app.Use(must(zip.Load(zip.Plugin{Name: "demo" + n, Bin: bin, Env: []string{
|
||||
"TESTPLUGIN_ON_TERM=" + filepath.Join(dir, n), "TESTPLUGIN_TERM_DELAY=600ms",
|
||||
}}, "/v1/demo"+n)))
|
||||
}
|
||||
for i := range 3 {
|
||||
if pidOf(app, "demo"+strconv.Itoa(i)) == 0 {
|
||||
t.Fatalf("demo%d is not running", i)
|
||||
}
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
if err := app.Shutdown(); err != nil {
|
||||
t.Fatalf("Shutdown: %v", err)
|
||||
}
|
||||
if took := time.Since(start); took > 1500*time.Millisecond {
|
||||
t.Fatalf("Shutdown took %s; three 600ms drains ran one after another", took)
|
||||
}
|
||||
for i := range 3 {
|
||||
if _, err := os.Stat(filepath.Join(dir, strconv.Itoa(i))); err != nil {
|
||||
t.Fatalf("child %d did not finish its shutdown: %v", i, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user