319 lines
8.7 KiB
Go
319 lines
8.7 KiB
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"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type helperTarget struct {
|
|
mu sync.Mutex
|
|
keys []Operation
|
|
failApply, failClose bool
|
|
closeCount int
|
|
applied, closed, blocked chan struct{}
|
|
gate chan struct{}
|
|
applyOnce, closeOnce sync.Once
|
|
}
|
|
|
|
func (target *helperTarget) Apply(name string, pressed bool) error {
|
|
target.mu.Lock()
|
|
kind := Up
|
|
if pressed {
|
|
kind = Down
|
|
}
|
|
target.keys = append(target.keys, Operation{Kind: kind, Name: name})
|
|
gate, fail := target.gate, target.failApply
|
|
target.mu.Unlock()
|
|
target.applyOnce.Do(func() { close(target.applied) })
|
|
if gate != nil {
|
|
close(target.blocked)
|
|
<-gate
|
|
}
|
|
if fail {
|
|
return errors.New("private target input")
|
|
}
|
|
return nil
|
|
}
|
|
func (target *helperTarget) Close() error {
|
|
target.mu.Lock()
|
|
target.closeCount++
|
|
fail := target.failClose
|
|
target.mu.Unlock()
|
|
target.closeOnce.Do(func() { close(target.closed) })
|
|
if fail {
|
|
return errors.New("private cleanup diagnostics")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type helperFixture struct {
|
|
control, life *io.PipeWriter
|
|
output *io.PipeReader
|
|
target *helperTarget
|
|
cancel context.CancelFunc
|
|
done chan struct{}
|
|
err error
|
|
}
|
|
|
|
func startHelper(t *testing.T, timeout time.Duration) *helperFixture {
|
|
t.Helper()
|
|
reader, control := io.Pipe()
|
|
lifetime, life := io.Pipe()
|
|
output, writer := io.Pipe()
|
|
ctx, cancel := context.WithCancel(t.Context())
|
|
target := &helperTarget{applied: make(chan struct{}), closed: make(chan struct{}), blocked: make(chan struct{})}
|
|
f := &helperFixture{control: control, life: life, output: output, target: target, cancel: cancel, done: make(chan struct{})}
|
|
go func() {
|
|
f.err = run(ctx, 7, reader, lifetime, writer, target, timeout)
|
|
close(f.done)
|
|
}()
|
|
t.Cleanup(func() {
|
|
cancel()
|
|
control.Close()
|
|
life.Close()
|
|
output.Close()
|
|
select {
|
|
case <-f.done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Error("helper I/O workers did not join")
|
|
}
|
|
})
|
|
return f
|
|
}
|
|
|
|
func (f *helperFixture) ack(t *testing.T) Ack {
|
|
t.Helper()
|
|
result := make(chan Ack, 1)
|
|
go func() {
|
|
frame := make([]byte, AckSize)
|
|
if _, err := io.ReadFull(f.output, frame); err != nil {
|
|
result <- Ack{}
|
|
return
|
|
}
|
|
ack, _ := DecodeAck(frame)
|
|
result <- ack
|
|
}()
|
|
select {
|
|
case ack := <-result:
|
|
if ack.Kind == 0 {
|
|
t.Fatal("invalid/missing helper acknowledgement")
|
|
}
|
|
return ack
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("helper acknowledgement timed out")
|
|
return Ack{}
|
|
}
|
|
}
|
|
|
|
func (f *helperFixture) send(t *testing.T, operation Operation) {
|
|
t.Helper()
|
|
frame, err := EncodeOperation(operation)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := writeAll(f.control, frame); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func (f *helperFixture) finish(t *testing.T, want error) {
|
|
t.Helper()
|
|
select {
|
|
case <-f.done:
|
|
if !errors.Is(f.err, want) {
|
|
t.Fatalf("helper result = %v; want %v", f.err, want)
|
|
}
|
|
f.target.mu.Lock()
|
|
defer f.target.mu.Unlock()
|
|
if f.target.closeCount != 1 {
|
|
t.Fatal("target cleanup was not performed exactly once")
|
|
}
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("helper cleanup did not finish")
|
|
}
|
|
}
|
|
|
|
func TestHelperRunnerOrderedClose(t *testing.T) {
|
|
f := startHelper(t, time.Second)
|
|
if ack := f.ack(t); ack.Kind != Ready || ack.Epoch != 7 || ack.Sequence != 0 {
|
|
t.Fatal("wrong helper readiness identity")
|
|
}
|
|
for index, kind := range []byte{Down, Heartbeat, Up, Close} {
|
|
operation := Operation{Kind: kind, Epoch: 7, Sequence: uint64(index + 1)}
|
|
if kind == Down || kind == Up {
|
|
operation.Name = "Control_R"
|
|
}
|
|
if index == 0 {
|
|
frame, _ := EncodeOperation(operation)
|
|
if err := writeAll(f.control, frame[:7]); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := writeAll(f.control, frame[7:]); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
} else {
|
|
f.send(t, operation)
|
|
}
|
|
ack := f.ack(t)
|
|
if ack.Sequence != operation.Sequence || ack.Epoch != 7 || ack.CleanupFailed {
|
|
t.Fatal("helper lost epoch, sequence or cleanup identity")
|
|
}
|
|
if kind == Close {
|
|
if ack.Kind != Closed || ack.Reason != ReasonClose {
|
|
t.Fatal("close was acknowledged before cleanup")
|
|
}
|
|
} else if ack.Kind != Accepted {
|
|
t.Fatal("valid operation was not accepted")
|
|
}
|
|
}
|
|
f.finish(t, nil)
|
|
if len(f.target.keys) != 2 || f.target.keys[0].Kind != Down || f.target.keys[1].Kind != Up {
|
|
t.Fatal("helper changed input order or synthesized keys")
|
|
}
|
|
}
|
|
|
|
func TestHelperRunnerRejectsStaleAndMalformed(t *testing.T) {
|
|
for _, failure := range []string{"epoch", "replay", "partial", "reserved", "lifetime-data"} {
|
|
t.Run(failure, func(t *testing.T) {
|
|
f := startHelper(t, time.Second)
|
|
f.ack(t)
|
|
f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "Shift_L"})
|
|
f.ack(t)
|
|
op := Operation{Kind: Heartbeat, Epoch: 7, Sequence: 2}
|
|
switch failure {
|
|
case "epoch":
|
|
op.Epoch++
|
|
f.send(t, op)
|
|
case "replay":
|
|
op.Sequence = 1
|
|
f.send(t, op)
|
|
case "partial":
|
|
if _, err := f.control.Write([]byte{Version}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
f.control.Close()
|
|
case "reserved":
|
|
frame, _ := EncodeOperation(op)
|
|
frame[3] = 1
|
|
if err := writeAll(f.control, frame); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
case "lifetime-data":
|
|
if _, err := f.life.Write([]byte{1}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if ack := f.ack(t); ack.Kind != Closed || ack.Reason != ReasonProtocol || ack.CleanupFailed {
|
|
t.Fatal("invalid input did not retire with cleanup", ack)
|
|
}
|
|
f.finish(t, ErrProtocol)
|
|
if len(f.target.keys) != 1 {
|
|
t.Fatal("invalid input reached the target")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestHelperRunnerLifetimeEOFCancelsPartialControl(t *testing.T) {
|
|
f := startHelper(t, time.Second)
|
|
f.ack(t)
|
|
f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "Shift_L"})
|
|
f.ack(t)
|
|
if _, err := f.control.Write([]byte{Version}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
f.life.Close() // Control remains open with an incomplete frame.
|
|
if ack := f.ack(t); ack.Kind != Closed || ack.Reason != ReasonEOF || ack.CleanupFailed {
|
|
t.Fatal("lifeline EOF waited for incomplete control data", ack)
|
|
}
|
|
f.finish(t, nil)
|
|
}
|
|
|
|
func TestHelperRunnerCancellationDiscardsQueuedDown(t *testing.T) {
|
|
f := startHelper(t, time.Second)
|
|
f.ack(t)
|
|
gate := make(chan struct{})
|
|
f.target.mu.Lock()
|
|
f.target.gate = gate
|
|
f.target.mu.Unlock()
|
|
f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "Shift_L"})
|
|
<-f.target.blocked
|
|
f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 2, Name: "a"})
|
|
f.cancel() // Synchronously invalidates the owner's context.
|
|
close(gate)
|
|
if ack := f.ack(t); ack.Kind != Closed || ack.Reason != ReasonCancelled {
|
|
t.Fatal("cancelled key was acknowledged")
|
|
}
|
|
f.finish(t, context.Canceled)
|
|
if len(f.target.keys) != 1 || f.target.keys[0].Name != "Shift_L" {
|
|
t.Fatal("queued down ran after cancellation")
|
|
}
|
|
}
|
|
|
|
func TestHelperRunnerBlockedAcksDoNotPostponeWatchdog(t *testing.T) {
|
|
f := startHelper(t, 50*time.Millisecond)
|
|
// Leave Ready unread, blocking the writer before it can write any ACK.
|
|
f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "Shift_L"})
|
|
select {
|
|
case <-f.target.closed:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("blocked output postponed native cleanup")
|
|
}
|
|
f.finish(t, ErrExpired)
|
|
}
|
|
|
|
func TestHelperRunnerKeyTrafficDoesNotRenewWatchdog(t *testing.T) {
|
|
f := startHelper(t, 100*time.Millisecond)
|
|
f.ack(t)
|
|
deadline := time.Now().Add(time.Second)
|
|
for sequence := uint64(1); ; sequence++ {
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("key-only traffic renewed helper authority")
|
|
}
|
|
frame, _ := EncodeOperation(Operation{Kind: Down, Epoch: 7, Sequence: sequence, Name: "a"})
|
|
err := writeAll(f.control, frame)
|
|
ack := f.ack(t)
|
|
if ack.Kind == Closed {
|
|
if ack.Reason != ReasonExpired {
|
|
t.Fatal("key-only traffic stopped for an unexpected reason", ack)
|
|
}
|
|
break
|
|
}
|
|
if err != nil || ack.Kind != Accepted {
|
|
t.Fatal("unexpected key-only acknowledgement", err, ack)
|
|
}
|
|
}
|
|
f.finish(t, ErrExpired)
|
|
}
|
|
|
|
func TestHelperRunnerTargetAndCleanupErrorsArePrivate(t *testing.T) {
|
|
for _, cleanupFailure := range []bool{false, true} {
|
|
f := startHelper(t, time.Second)
|
|
f.ack(t)
|
|
f.target.mu.Lock()
|
|
f.target.failApply, f.target.failClose = true, cleanupFailure
|
|
f.target.mu.Unlock()
|
|
f.send(t, Operation{Kind: Down, Epoch: 7, Sequence: 1, Name: "a"})
|
|
if ack := f.ack(t); ack.Kind != Closed || ack.Reason != ReasonTarget || ack.CleanupFailed != cleanupFailure {
|
|
t.Fatal("native failure lost its cleanup status", ack)
|
|
}
|
|
want := ErrTarget
|
|
if cleanupFailure {
|
|
want = ErrCleanup
|
|
}
|
|
f.finish(t, want)
|
|
if strings.Contains(f.err.Error(), "private") {
|
|
t.Fatal("target error leaked input or diagnostics")
|
|
}
|
|
}
|
|
}
|