cthulhu/internal/remoteinput/supervisor.go

// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
// SPDX-License-Identifier: GPL-3.0-or-later

package remoteinput

import (
	"context"
	"errors"
	"io"
	"math"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"sync"
	"syscall"
	"time"
)

var (
	ErrUnavailable = errors.New("remote input helper unavailable")
	ErrUncertain   = errors.New("remote input helper retirement unconfirmed")
)

const operationTimeout = time.Second
const retirementTimeout = Watchdog + time.Second

// Supervisor is the reader-side helper process owner. It grants no consent.
// Its production owner binds ctx/epoch to an approved lease and rechecks
// native policy before every Apply. Unconfirmed cleanup permanently blocks
// replacement on this Supervisor.
type Supervisor struct {
	mu        sync.Mutex
	active    bool
	uncertain bool
	command   func(uint64) *exec.Cmd
}

func NewSupervisor(executable string) (*Supervisor, error) {
	if !filepath.IsAbs(executable) {
		return nil, ErrUnavailable
	}
	return &Supervisor{command: func(epoch uint64) *exec.Cmd {
		return exec.Command(executable, "--epoch", strconv.FormatUint(epoch, 10))
	}}, nil
}

type keyCall struct {
	name    string
	pressed bool
	result  chan error
}

// Client owns one fixed epoch. Cancel closes its independent lifetime pipe;
// Done closes only after retirement was checked or its bounded wait expired.
// Cancellation never kills the helper: it must remain alive to release keys.
type Client struct {
	cancel context.CancelFunc
	ready  chan struct{}
	done   chan struct{}
	calls  chan keyCall
	err    error // published by closing done
}

func (supervisor *Supervisor) Start(ctx context.Context, epoch uint64) (*Client, error) {
	if ctx == nil || ctx.Err() != nil || epoch == 0 {
		return nil, ErrUnavailable
	}
	supervisor.mu.Lock()
	if supervisor.uncertain || supervisor.active {
		supervisor.mu.Unlock()
		return nil, ErrUnavailable
	}
	supervisor.active = true
	supervisor.mu.Unlock()
	owner, cancel := context.WithCancel(ctx)
	client := &Client{cancel: cancel, ready: make(chan struct{}), done: make(chan struct{}), calls: make(chan keyCall)}
	go func() {
		err := supervisor.serve(owner, epoch, client)
		cancel()
		supervisor.mu.Lock()
		supervisor.active = false
		supervisor.uncertain = errors.Is(err, ErrUncertain)
		supervisor.mu.Unlock()
		client.err = err
		close(client.done)
	}()
	select {
	case <-client.ready:
		if owner.Err() == nil {
			return client, nil
		}
	case <-client.done:
		return nil, client.err
	case <-owner.Done():
	}
	cancel()
	<-client.done
	if client.err != nil {
		return nil, client.err
	}
	return nil, ErrUnavailable
}

func (client *Client) Cancel()               { client.cancel() }
func (client *Client) Done() <-chan struct{} { return client.done }

// Wait reports cleanup certainty, not whether the last key reached an app.
func (client *Client) Wait() error {
	<-client.done
	return client.err
}

// Apply has no retained queue. Cancellation after admission retires the entire
// client rather than allowing an uncertain key transition to outlive its call.
func (client *Client) Apply(ctx context.Context, name string, pressed bool) error {
	if ctx == nil || ctx.Err() != nil || !validKey(name) {
		return ErrUnavailable
	}
	call := keyCall{name: name, pressed: pressed, result: make(chan error, 1)}
	select {
	case <-client.done:
		return ErrUnavailable
	case <-ctx.Done():
		return ErrUnavailable
	case client.calls <- call:
	}
	select {
	case err := <-call.result:
		return err
	case <-client.done:
		return ErrUnavailable
	case <-ctx.Done():
		client.Cancel()
		return ErrUnavailable
	}
}

func validKey(name string) bool {
	_, err := EncodeOperation(Operation{Kind: Down, Epoch: 1, Sequence: 1, Name: name})
	return err == nil
}

type ackResult struct {
	ack Ack
	err error
}

func (supervisor *Supervisor) serve(ctx context.Context, epoch uint64, client *Client) error {
	var files []*os.File
	defer func() {
		for _, file := range files {
			file.Close()
		}
	}()
	pipe := func() (*os.File, *os.File, error) {
		read, write, err := os.Pipe()
		if err == nil {
			files = append(files, read, write)
		}
		return read, write, err
	}
	controlRead, control, err := pipe()
	if err != nil {
		return ErrUnavailable
	}
	output, outputWrite, err := pipe()
	if err != nil {
		return ErrUnavailable
	}
	lifetimeRead, lifetime, err := pipe()
	if err != nil {
		return ErrUnavailable
	}
	cmd := supervisor.command(epoch)
	cmd.Stdin, cmd.Stdout, cmd.Stderr = controlRead, outputWrite, io.Discard
	cmd.ExtraFiles = []*os.File{lifetimeRead}
	// No CommandContext/Pdeathsig: writer death must allow native retirement.
	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	if ctx.Err() != nil || cmd.Start() != nil {
		return ErrUnavailable
	}
	controlRead.Close()
	outputWrite.Close()
	lifetimeRead.Close()
	process := make(chan error, 1)
	go func() { process <- cmd.Wait() }()
	stopWorkers := make(chan struct{})
	var workers sync.WaitGroup
	workers.Add(2)
	defer func() {
		close(stopWorkers)
		output.Close()
		workers.Wait()
	}()
	go func() {
		defer workers.Done()
		select {
		case <-ctx.Done():
			lifetime.Close()
		case <-stopWorkers:
		}
	}()
	acks := make(chan ackResult, 1)
	go func() {
		defer workers.Done()
		for {
			var frame [AckSize]byte
			_, err := io.ReadFull(output, frame[:])
			var ack Ack
			if err == nil {
				ack, err = DecodeAck(frame[:])
			}
			select {
			case acks <- ackResult{ack, err}:
			case <-stopWorkers:
				return
			}
			if err != nil {
				return
			}
		}
	}()
	var sent, accepted uint64
	var terminal *Ack
	protocolFailed := false
	streamEnded, readySeen := false, false
	// Every return after launch goes through the independent cleanup barrier.
	retire := func() error {
		lifetime.Close()
		control.Close()
		timer := time.NewTimer(retirementTimeout)
		defer timer.Stop()
		var exited bool
		var exitError error
		for !exited || !streamEnded {
			select {
			case exitError = <-process:
				exited = true
			case result := <-acks:
				if result.err != nil {
					streamEnded = true
					if result.err != io.EOF {
						protocolFailed = true
					}
					continue
				}
				ack := result.ack
				if terminal != nil || ack.Epoch != epoch {
					protocolFailed = true
				} else if !readySeen && ack == (Ack{Kind: Ready, Epoch: epoch}) {
					readySeen = true
				} else if ack.Kind == Accepted && ack.Sequence == sent && accepted < sent {
					accepted = sent
				} else if ack.Kind == Closed && ack.Sequence >= accepted && ack.Sequence <= sent {
					terminal = &ack
				} else {
					protocolFailed = true
				}
			case <-timer.C:
				// A blocked native server may leave a living helper. Never kill
				// it or permit replacement; its eventual exit is still reaped.
				return ErrUncertain
			}
		}
		if protocolFailed || terminal == nil || terminal.CleanupFailed || exitError != nil {
			return ErrUncertain
		}
		return nil
	}
	readAck := func() (Ack, error) {
		timer := time.NewTimer(operationTimeout)
		defer timer.Stop()
		select {
		case result := <-acks:
			if result.err != nil {
				streamEnded = true
				if result.err != io.EOF {
					protocolFailed = true
				}
			}
			return result.ack, result.err
		case <-ctx.Done():
			return Ack{}, ErrUnavailable
		case <-timer.C:
			return Ack{}, ErrUnavailable
		}
	}
	ready, err := readAck()
	if err != nil || ready != (Ack{Kind: Ready, Epoch: epoch}) {
		// An ACK consumed here cannot be reinterpreted as a valid terminal.
		if err == nil {
			protocolFailed = true
		}
		if cleanup := retire(); cleanup != nil {
			return cleanup
		}
		return ErrUnavailable
	}
	readySeen = true
	close(client.ready)
	transact := func(kind byte, name string) error {
		if ctx.Err() != nil || sent == math.MaxUint64 {
			return ErrUnavailable
		}
		sent++
		frame, err := EncodeOperation(Operation{Kind: kind, Epoch: epoch, Sequence: sent, Name: name})
		if err != nil {
			return ErrUnavailable
		}
		if err = control.SetWriteDeadline(time.Now().Add(operationTimeout)); err == nil {
			err = writeAll(control, frame)
		}
		if err != nil {
			return ErrUnavailable
		}
		ack, err := readAck()
		if err != nil {
			return ErrUnavailable
		}
		if ack == (Ack{Kind: Accepted, Epoch: epoch, Sequence: sent}) {
			accepted = sent
			return nil
		}
		if ack.Kind == Closed && ack.Epoch == epoch && ack.Sequence >= accepted && ack.Sequence <= sent {
			terminal = &ack
		} else {
			protocolFailed = true
		}
		return ErrUnavailable
	}
	ticker := time.NewTicker(Watchdog / 3)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return retire()
		case <-ticker.C:
			if transact(Heartbeat, "") != nil {
				return retire()
			}
		case call := <-client.calls:
			kind := Up
			if call.pressed {
				kind = Down
			}
			err := transact(kind, call.name)
			call.result <- err
			if err != nil {
				return retire()
			}
		case result := <-acks:
			if result.err != nil {
				streamEnded = true
				protocolFailed = true
			} else if result.ack.Kind == Closed && result.ack.Epoch == epoch && result.ack.Sequence == accepted {
				terminal = &result.ack
			} else {
				protocolFailed = true
			}
			return retire()
		}
	}
}