// 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 (
"errors"
"log"
"os/exec"
"sync"
"syscall"
"time"
)
// ErrStopping is Track's answer while a stop is waiting for children:
// 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
cmd *exec.Cmd
start time.Time
}
// New returns an empty set that logs through logf.
func New(logf func(string, ...any)) *Set {
return &Set{logf: logf, live: map[*entry]struct{}{}}
}
// Track registers cmd, built but not necessarily started, as a child
// serving repo. The caller calls done once the process has been waited
// for (or was never started). While a stop is waiting it refuses with
// ErrStopping, and the caller must not start cmd.
func (s *Set) Track(repo, verb string, cmd *exec.Cmd) (done func(), err error) {
e := &entry{repo: repo, verb: verb, cmd: cmd, start: time.Now()}
s.mu.Lock()
if s.closed {
s.mu.Unlock()
return func() {}, ErrStopping
}
s.live[e] = struct{}{}
s.mu.Unlock()
var once sync.Once
return func() { once.Do(func() { s.ended(e) }) }, nil
}
func (s *Set) ended(e *entry) {
s.mu.Lock()
delete(s.live, e)
stopping := s.stopping
s.mu.Unlock()
if stopping {
s.logf("stopping: zit child ended: repo=%s verb=%s pid=%d exit=%s took=%v",
e.repo, e.verb, pid(e.cmd), exit(e.cmd), time.Since(e.start).Round(time.Millisecond))
}
}
// 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. New children are refused while it waits.
// It returns how many were left running.
//
// The set takes children again once Wait returns. In production the
// process exits next; a test that runs a stop keeps a usable set.
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()
defer func() {
s.mu.Lock()
s.stopping, s.closed = false, false
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 pid=%d for %v",
e.repo, e.verb, pid(e.cmd), time.Since(e.start).Round(time.Millisecond))
}
if s.until(bound) {
return 0
}
s.mu.Lock()
survivors := s.snapshot()
s.mu.Unlock()
for _, e := range survivors {
s.logf("stopping: zit child outlived the wait; sending SIGTERM: repo=%s verb=%s pid=%d",
e.repo, e.verb, pid(e.cmd))
if e.cmd.Process != nil {
_ = e.cmd.Process.Signal(syscall.SIGTERM)
}
}
if s.until(grace) {
return 0
}
s.mu.Lock()
survivors = s.snapshot()
s.mu.Unlock()
for _, e := range survivors {
s.logf("stopping: zit child still running at exit: repo=%s verb=%s pid=%d for %v",
e.repo, e.verb, pid(e.cmd), time.Since(e.start).Round(time.Millisecond))
}
return len(survivors)
}
// 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
}
func pid(cmd *exec.Cmd) int {
if cmd.Process == nil {
return 0
}
return cmd.Process.Pid
}
func exit(cmd *exec.Cmd) string {
switch {
case cmd.Process == nil:
return "never started"
case cmd.ProcessState == nil:
return "not waited for"
default:
return cmd.ProcessState.String()
}
}
// Default is the process's own set, logging with the standard logger.
var Default = New(log.Printf)