// Package children tracks the subprocesses the forge starts, so a stop
// can wait for them and say how each one ended.
//
// Under systemd's KillMode=mixed only quarry is sent SIGTERM; anything
// it leaves running when it exits is SIGKILLed. A stop that returned
// while a store command or a fronted `zit serve` was still running was
// therefore worse than no drain at all. Every zit child is registered
// here where it is built, and the stop waits on this set before the
// process exits.
//
// A leaf package because children are started in several places (the
// store client, the sync fronter) that must not import each other.
package children
import (
"context"
"errors"
"log"
"os/exec"
"sync"
"syscall"
"time"
)
// ErrStopping is the set's answer once a stop has begun waiting: a
// child started then would only extend the wait it is part of.
var ErrStopping = errors.New("the forge is stopping; no new subprocesses")
// Set is the children of one process.
type Set struct {
logf func(string, ...any)
mu sync.Mutex
live map[*entry]struct{}
stopping bool // log each child as it ends
closed bool // refuse new children
}
type entry struct {
repo, verb string
start time.Time
cancel context.CancelFunc
// Written only under Set.mu. The set never reads the command's
// Process field itself: the caller's Start writes it with no lock
// the set could share, so a stop reading it would race a spawn.
terminating bool // the stop asked for SIGTERM rather than a kill
pid int // recorded by the Cancel hook, where it is safe to read
}
// New returns an empty set that logs through logf.
func New(logf func(string, ...any)) *Set {
return &Set{logf: logf, live: map[*entry]struct{}{}}
}
// Command builds name args as a child serving repo and registers it.
// The caller runs it (Start, Run or Output) and calls done once it has
// been waited for, or was never started, on the goroutine that ran it.
//
// ctx cancels it as exec.CommandContext does, with a kill. The set can
// also end it: when a stop's wait runs out, a child still running is
// sent SIGTERM through the same Cancel hook, which is the one place
// os/exec guarantees the process is set and safe to read.
//
// Once a stop has begun waiting it refuses with ErrStopping, and the
// returned command carries that error, so running it fails before
// anything is spawned.
func (s *Set) Command(ctx context.Context, repo, verb, name string, args ...string) (cmd *exec.Cmd, done func(), err error) {
if ctx == nil {
ctx = context.Background()
}
cctx, cancel := context.WithCancel(ctx)
cmd = exec.CommandContext(cctx, name, args...)
e := &entry{repo: repo, verb: verb, start: time.Now(), cancel: cancel}
cmd.Cancel = func() error {
s.mu.Lock()
e.pid = cmd.Process.Pid
terminating := e.terminating
s.mu.Unlock()
if terminating {
s.logf("stopping: zit child outlived the wait; sending SIGTERM: repo=%s verb=%s pid=%d",
e.repo, e.verb, e.pid)
return cmd.Process.Signal(syscall.SIGTERM)
}
return cmd.Process.Kill()
}
s.mu.Lock()
if s.closed {
s.mu.Unlock()
cancel()
cmd.Err = ErrStopping
return cmd, func() {}, ErrStopping
}
s.live[e] = struct{}{}
s.mu.Unlock()
var once sync.Once
return cmd, func() { once.Do(func() { s.ended(e, cmd) }) }, nil
}
// ended runs on the goroutine that ran cmd, after its Wait, so reading
// the command's process state here is ordered after Start wrote it. The
// line is logged before the entry leaves the set, so a stop that sees
// the set empty has already logged every end.
func (s *Set) ended(e *entry, cmd *exec.Cmd) {
s.mu.Lock()
stopping := s.stopping
s.mu.Unlock()
if stopping {
pid, exit := 0, "never started"
if cmd.Process != nil {
pid = cmd.Process.Pid
exit = "not waited for"
if cmd.ProcessState != nil {
exit = cmd.ProcessState.String()
}
}
s.logf("stopping: zit child ended: repo=%s verb=%s pid=%d exit=%s took=%v",
e.repo, e.verb, pid, exit, time.Since(e.start).Round(time.Millisecond))
}
s.mu.Lock()
delete(s.live, e)
s.mu.Unlock()
// The child has been waited for, so cancelling kills nothing; it
// releases the per-child context from a long-lived parent at once
// instead of when that parent ends.
e.cancel()
}
// Stopping starts logging each child as it ends, so the stop's log
// shows which children ended during the drain and which after it.
func (s *Set) Stopping() {
s.mu.Lock()
s.stopping = true
s.mu.Unlock()
}
// Live is how many children are running or about to.
func (s *Set) Live() int {
s.mu.Lock()
defer s.mu.Unlock()
return len(s.live)
}
// Wait holds a stop until every child has ended, for up to bound. The
// survivors are then sent SIGTERM and given grace to go; any still
// running after that is logged by name, since systemd will SIGKILL it
// when this process exits. It returns how many were left running.
//
// From the moment it is called the set refuses new children, and it
// stays that way: the process exits next, and a child started in the
// meantime would be SIGKILLed with nobody waiting for it. Reset exists
// for tests that run a stop in-process.
func (s *Set) Wait(bound, grace time.Duration) (left int) {
s.mu.Lock()
s.stopping, s.closed = true, true
waiting := s.snapshot()
s.mu.Unlock()
if len(waiting) == 0 {
return 0
}
s.logf("stopping: waiting up to %v for %d zit children", bound, len(waiting))
for _, e := range waiting {
s.logf("stopping: zit child running: repo=%s verb=%s for %v",
e.repo, e.verb, time.Since(e.start).Round(time.Millisecond))
}
if s.until(bound) {
return 0
}
s.mu.Lock()
survivors := s.snapshot()
for _, e := range survivors {
e.terminating = true
}
s.mu.Unlock()
// Cancelling runs each command's Cancel hook, which sends SIGTERM
// now that terminating is set; a child not yet started never will be.
for _, e := range survivors {
e.cancel()
}
if s.until(grace) {
return 0
}
s.mu.Lock()
survivors = s.snapshot()
type still struct {
repo, verb string
pid int
age time.Duration
}
var named []still
for _, e := range survivors {
named = append(named, still{e.repo, e.verb, e.pid, time.Since(e.start).Round(time.Millisecond)})
}
s.mu.Unlock()
for _, n := range named {
s.logf("stopping: zit child still running at exit: repo=%s verb=%s pid=%d for %v",
n.repo, n.verb, n.pid, n.age)
}
return len(named)
}
// Reset reopens a set a stop has closed, for a test that ran a stop
// in-process and goes on to start children. Production never calls it.
func (s *Set) Reset() {
s.mu.Lock()
s.stopping, s.closed = false, false
s.mu.Unlock()
}
// until polls for an empty set; children end on their own schedule and
// the wait is short, so a poll is simpler than a condition per child.
func (s *Set) until(bound time.Duration) bool {
deadline := time.Now().Add(bound)
for {
if s.Live() == 0 {
return true
}
if !time.Now().Before(deadline) {
return false
}
time.Sleep(10 * time.Millisecond)
}
}
func (s *Set) snapshot() []*entry {
out := make([]*entry, 0, len(s.live))
for e := range s.live {
out = append(out, e)
}
return out
}
// Default is the process's own set, logging with the standard logger.
var Default = New(log.Printf)