cthulhu/internal/remoteinput/runner.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"
	"sync"
	"time"
)

const Watchdog = 3 * time.Second
const ackQueueSize = 32
const finalAckTimeout = 250 * time.Millisecond

var (
	ErrExpired = errors.New("remote input helper heartbeat expired")
	ErrTarget  = errors.New("remote input helper target failed")
	ErrOutput  = errors.New("remote input helper acknowledgements unavailable")
	ErrCleanup = errors.New("remote input helper cleanup failed")
	errEOF     = errors.New("remote input helper owner closed")
)

// Target methods run on Run's caller, never on its I/O goroutines. Close must
// retire keys before destroying native state. A blocked native method cannot
// be interrupted here: watchdog recovery requires a responsive native server.
type Target interface {
	Apply(name string, pressed bool) error
	Close() error
}

type ackWrite struct {
	ack  Ack
	done chan error
}

// Run owns valid supplied streams and target through cleanup and I/O join.
// Stream Close must interrupt blocked reads/writes; AdoptPipe supplies that
// property for inherited Go file descriptors. The caller supplies one fixed
// nonzero epoch and must not launch a replacement until cleanup is confirmed.
func Run(ctx context.Context, epoch uint64, control, lifetime io.ReadCloser, output io.WriteCloser, target Target) error {
	return run(ctx, epoch, control, lifetime, output, target, Watchdog)
}

func run(ctx context.Context, epoch uint64, control, lifetime io.ReadCloser, output io.WriteCloser, target Target, timeout time.Duration) error {
	if ctx == nil || epoch == 0 || control == nil || lifetime == nil || output == nil || target == nil || timeout <= 0 {
		return ErrProtocol
	}
	ownerContext, cancelOwner := context.WithCancelCause(ctx)
	defer cancelOwner(context.Canceled)
	ioContext, stopIO := context.WithCancel(context.Background())
	results := make(chan Operation, 1)
	acks := make(chan ackWrite, ackQueueSize)
	writeFailure := make(chan struct{}, 1)
	var workers sync.WaitGroup
	workers.Add(3)
	go func() {
		defer workers.Done()
		// This pipe never carries data. Its EOF cannot sit behind queued keys.
		var value [1]byte
		n, err := lifetime.Read(value[:])
		if n == 0 && err == io.EOF {
			cancelOwner(errEOF)
		} else {
			cancelOwner(ErrProtocol)
		}
	}()
	go func() {
		defer workers.Done()
		for {
			frame := make([]byte, OperationSize)
			_, err := io.ReadFull(control, frame)
			var operation Operation
			if err == nil {
				operation, err = DecodeOperation(frame)
			}
			if err != nil {
				if err == io.EOF {
					cancelOwner(errEOF)
				} else {
					cancelOwner(ErrProtocol)
				}
				return
			}
			select {
			case results <- operation:
			case <-ioContext.Done():
				return
			}
		}
	}()
	go func() {
		defer workers.Done()
		for {
			select {
			case <-ioContext.Done():
				return
			case item := <-acks:
				if ioContext.Err() != nil {
					return
				}
				frame, err := EncodeAck(item.ack)
				if err == nil {
					err = writeAll(output, frame)
				}
				if item.done != nil {
					item.done <- err
				}
				if err != nil {
					cancelOwner(ErrOutput)
					writeFailure <- struct{}{}
					return
				}
			}
		}
	}()
	defer func() {
		stopIO()
		_ = control.Close()
		_ = lifetime.Close()
		_ = output.Close()
		workers.Wait()
	}()
	enqueue := func(ack Ack, done chan error) bool {
		select {
		case acks <- ackWrite{ack: ack, done: done}:
			return true
		default:
			return false
		}
	}
	deadline := time.Now().Add(timeout)
	timer := time.NewTimer(timeout)
	defer timer.Stop()
	var sequence uint64
	var cause error
	reason := ReasonEOF
	ownerStopped := func() {
		cause = context.Cause(ownerContext)
		switch cause {
		case errEOF:
			cause, reason = nil, ReasonEOF
		case ErrProtocol:
			reason = ReasonProtocol
		case ErrOutput:
			reason = ReasonOutput
		default:
			cause = ctx.Err()
			if cause == nil {
				cause = context.Canceled
			}
			reason = ReasonCancelled
		}
	}
	enqueue(Ack{Kind: Ready, Epoch: epoch}, nil)
loop:
	for {
		select {
		case <-ownerContext.Done():
			ownerStopped()
			break loop
		case <-writeFailure:
			cause, reason = ErrOutput, ReasonOutput
			break loop
		case <-timer.C:
			if time.Now().Before(deadline) {
				timer.Reset(time.Until(deadline))
				continue
			}
			cause, reason = ErrExpired, ReasonExpired
			break loop
		case operation := <-results:
			// Check before every frame, including a heartbeat. Busy input or a
			// late heartbeat cannot win a select race against expired authority.
			if ownerContext.Err() != nil {
				ownerStopped()
				break loop
			}
			if !time.Now().Before(deadline) {
				cause, reason = ErrExpired, ReasonExpired
				break loop
			}
			if operation.Epoch != epoch || operation.Sequence <= sequence {
				cause, reason = ErrProtocol, ReasonProtocol
				break loop
			}
			sequence = operation.Sequence
			switch operation.Kind {
			case Close:
				reason = ReasonClose
				break loop
			case Heartbeat:
				deadline = time.Now().Add(timeout)
				timer.Reset(timeout)
			case Down, Up:
				if err := target.Apply(operation.Name, operation.Kind == Down); err != nil {
					cause, reason = ErrTarget, ReasonTarget
					break loop
				}
			}
			if ownerContext.Err() != nil {
				ownerStopped()
				break loop
			}
			if !time.Now().Before(deadline) {
				cause, reason = ErrExpired, ReasonExpired
				break loop
			}
			if !enqueue(Ack{Kind: Accepted, Epoch: epoch, Sequence: sequence}, nil) {
				cause, reason = ErrOutput, ReasonOutput
				break loop
			}
		}
	}
	// No acknowledgement or blocked pipe can postpone native cleanup.
	_ = control.Close()
	cleanup := target.Close()
	done := make(chan error, 1)
	if enqueue(Ack{Kind: Closed, Reason: reason, CleanupFailed: cleanup != nil, Epoch: epoch, Sequence: sequence}, done) {
		flush := time.NewTimer(finalAckTimeout)
		select {
		case err := <-done:
			if err != nil && cause == nil {
				cause = ErrOutput
			}
		case <-writeFailure:
			if cause == nil {
				cause = ErrOutput
			}
		case <-flush.C:
			if cause == nil {
				cause = ErrOutput
			}
		}
		flush.Stop()
	} else if cause == nil {
		cause = ErrOutput
	}
	if cleanup != nil {
		return ErrCleanup
	}
	return cause
}

func writeAll(writer io.Writer, frame []byte) error {
	for len(frame) > 0 {
		n, err := writer.Write(frame)
		if err != nil {
			return err
		}
		if n <= 0 || n > len(frame) {
			return io.ErrShortWrite
		}
		frame = frame[n:]
	}
	return nil
}