mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
feat: bound supervisor drain termination
This commit is contained in:
@@ -46,6 +46,8 @@ type Supervisor struct {
|
||||
cmd *exec.Cmd
|
||||
}
|
||||
|
||||
const DefaultDrainGrace = 285 * time.Second
|
||||
|
||||
func New(config Config) (*Supervisor, error) {
|
||||
if len(config.Command) == 0 || config.Command[0] == "" {
|
||||
return nil, fmt.Errorf("supervisor command is required")
|
||||
@@ -142,6 +144,58 @@ func (s *Supervisor) Wait() error {
|
||||
return s.cmd.Wait()
|
||||
}
|
||||
|
||||
// Run owns the PID-1 termination sequence. The child gets its own context so
|
||||
// cancellation of the supervisor does not kill it before the authenticated
|
||||
// drain request has had a chance to stop new admissions. A non-responsive
|
||||
// child is force-killed after drainGrace; a drain failure is recorded only by
|
||||
// the returned error if the child exits cleanly, while the deadline still
|
||||
// prevents a stuck process from hanging termination forever.
|
||||
func (s *Supervisor) Run(ctx context.Context, drainGrace time.Duration) error {
|
||||
if s == nil || ctx == nil {
|
||||
return fmt.Errorf("supervisor context is required")
|
||||
}
|
||||
if drainGrace <= 0 {
|
||||
drainGrace = DefaultDrainGrace
|
||||
}
|
||||
processCtx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
if err := s.Start(processCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
wait := make(chan error, 1)
|
||||
go func() { wait <- s.Wait() }()
|
||||
select {
|
||||
case err := <-wait:
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
}
|
||||
|
||||
var drainErr error
|
||||
if s.config.DrainURL != "" {
|
||||
drainCtx, drainCancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
drainErr = s.Drain(drainCtx)
|
||||
drainCancel()
|
||||
}
|
||||
timer := time.NewTimer(drainGrace)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case err := <-wait:
|
||||
if drainErr != nil {
|
||||
return fmt.Errorf("child exited after drain failure: %w", drainErr)
|
||||
}
|
||||
return err
|
||||
case <-timer.C:
|
||||
if s.cmd != nil && s.cmd.Process != nil {
|
||||
_ = s.cmd.Process.Kill()
|
||||
}
|
||||
<-wait
|
||||
if drainErr != nil {
|
||||
return fmt.Errorf("drain failed and child was force-killed: %w", drainErr)
|
||||
}
|
||||
return fmt.Errorf("child force-killed after drain deadline")
|
||||
}
|
||||
}
|
||||
|
||||
// Drain asks the allocated Godot process to stop accepting new work. The
|
||||
// token is sent only over the configured localhost control endpoint and is
|
||||
// never placed in command arguments or logs.
|
||||
|
||||
@@ -180,3 +180,53 @@ func TestSupervisorRejectsRemoteOrPartialDrainConfiguration(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunDrainsBeforeChildExit(t *testing.T) {
|
||||
marker := filepath.Join(t.TempDir(), "drained")
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/drain" {
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
if r.Header.Get("Authorization") != "Bearer run-secret" {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
if err := os.WriteFile(marker, []byte("drained"), 0600); err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusAccepted)
|
||||
}))
|
||||
defer server.Close()
|
||||
command := []string{"/bin/sh", "-c", "while [ ! -f '" + marker + "' ]; do sleep 0.01; done"}
|
||||
s, err := New(Config{Command: command, DrainURL: server.URL + "/drain", DrainToken: "run-secret"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
time.AfterFunc(30*time.Millisecond, cancel)
|
||||
if err := s.Run(ctx, time.Second); err != nil {
|
||||
t.Fatalf("graceful run: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(marker); err != nil {
|
||||
t.Fatalf("drain endpoint was not called: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunForceKillsUnresponsiveChildAtDeadline(t *testing.T) {
|
||||
s, err := New(Config{Command: []string{"/bin/sh", "-c", "trap '' TERM; sleep 5"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
time.AfterFunc(30*time.Millisecond, cancel)
|
||||
started := time.Now()
|
||||
err = s.Run(ctx, 50*time.Millisecond)
|
||||
if err == nil || !strings.Contains(err.Error(), "force-killed") {
|
||||
t.Fatalf("unresponsive child result = %v", err)
|
||||
}
|
||||
if elapsed := time.Since(started); elapsed > time.Second {
|
||||
t.Fatalf("force-kill exceeded bounded deadline: %s", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user