toki
fef1f7a9dc
Node의 provider progress 기반 stall timeout, watchdog fencing과 bounded health probe evidence를 실행 경로에 반영한다. Edge-Node 계약과 구현 스펙, 테스트 및 Milestone 완료 evidence를 현재 상태와 맞춘다.
790 lines
27 KiB
Go
790 lines
27 KiB
Go
package node
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
toki "git.toki-labs.com/toki/proto-socket/go"
|
|
"go.uber.org/zap"
|
|
|
|
"google.golang.org/protobuf/proto"
|
|
|
|
"iop/apps/node/internal/store"
|
|
"iop/apps/node/internal/transport"
|
|
runtime "iop/packages/go/execution"
|
|
iop "iop/proto/gen/iop"
|
|
)
|
|
|
|
type manualAttemptTimer struct {
|
|
mu sync.Mutex
|
|
ch chan time.Time
|
|
now func() time.Time
|
|
advanceTo func(time.Time)
|
|
durations []time.Duration
|
|
scheduled time.Time
|
|
stopped bool
|
|
fired bool
|
|
beforeReset func()
|
|
}
|
|
|
|
func newManualAttemptTimer(d time.Duration, now func() time.Time, advanceTo func(time.Time)) *manualAttemptTimer {
|
|
scheduled := now().Add(d)
|
|
return &manualAttemptTimer{ch: make(chan time.Time, 1), now: now, advanceTo: advanceTo, durations: []time.Duration{d}, scheduled: scheduled}
|
|
}
|
|
func (t *manualAttemptTimer) C() <-chan time.Time { return t.ch }
|
|
func (t *manualAttemptTimer) Stop() bool {
|
|
t.mu.Lock()
|
|
defer t.mu.Unlock()
|
|
wasActive := !t.stopped && !t.fired
|
|
t.stopped = true
|
|
return wasActive
|
|
}
|
|
func (t *manualAttemptTimer) Reset(d time.Duration) bool {
|
|
t.mu.Lock()
|
|
beforeReset := t.beforeReset
|
|
t.mu.Unlock()
|
|
if beforeReset != nil {
|
|
beforeReset()
|
|
}
|
|
t.mu.Lock()
|
|
defer t.mu.Unlock()
|
|
wasStopped := t.stopped
|
|
t.stopped = false
|
|
t.fired = false
|
|
t.durations = append(t.durations, d)
|
|
t.scheduled = t.now().Add(d)
|
|
return wasStopped
|
|
}
|
|
func (t *manualAttemptTimer) fire() {
|
|
t.mu.Lock()
|
|
stopped, fired, scheduled := t.stopped, t.fired, t.scheduled
|
|
if !stopped && !fired {
|
|
t.fired = true
|
|
}
|
|
t.mu.Unlock()
|
|
if !stopped && !fired {
|
|
t.advanceTo(scheduled)
|
|
t.ch <- scheduled
|
|
}
|
|
}
|
|
func (t *manualAttemptTimer) fireStaleArmDuringReset() {
|
|
t.mu.Lock()
|
|
scheduled := t.scheduled
|
|
t.mu.Unlock()
|
|
t.advanceTo(scheduled)
|
|
t.ch <- scheduled
|
|
}
|
|
func (t *manualAttemptTimer) snapshot() ([]time.Duration, bool) {
|
|
t.mu.Lock()
|
|
defer t.mu.Unlock()
|
|
return append([]time.Duration(nil), t.durations...), t.stopped
|
|
}
|
|
|
|
type manualAttemptClock struct {
|
|
mu sync.Mutex
|
|
timers []*manualAttemptTimer
|
|
created chan struct{}
|
|
now time.Time
|
|
beforeTimerReturn func(*manualAttemptTimer)
|
|
}
|
|
|
|
func newManualAttemptClock() *manualAttemptClock {
|
|
return &manualAttemptClock{created: make(chan struct{}, 16), now: time.Unix(0, 0)}
|
|
}
|
|
|
|
// Now returns a strictly increasing timestamp. Timers retain their scheduled
|
|
// deadline separately, so a delayed manual fire cannot be mistaken for the
|
|
// clock's later read time.
|
|
func (c *manualAttemptClock) Now() time.Time {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
c.now = c.now.Add(time.Millisecond)
|
|
return c.now
|
|
}
|
|
func (c *manualAttemptClock) current() time.Time {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return c.now
|
|
}
|
|
func (c *manualAttemptClock) advanceTo(at time.Time) {
|
|
c.mu.Lock()
|
|
if c.now.Before(at) {
|
|
c.now = at
|
|
}
|
|
c.mu.Unlock()
|
|
}
|
|
func (c *manualAttemptClock) NewTimer(d time.Duration) attemptTimer {
|
|
timer := newManualAttemptTimer(d, c.current, c.advanceTo)
|
|
c.mu.Lock()
|
|
c.timers = append(c.timers, timer)
|
|
beforeTimerReturn := c.beforeTimerReturn
|
|
c.mu.Unlock()
|
|
c.created <- struct{}{}
|
|
if beforeTimerReturn != nil {
|
|
beforeTimerReturn(timer)
|
|
}
|
|
return timer
|
|
}
|
|
func (c *manualAttemptClock) waitTimer(t *testing.T, index int) *manualAttemptTimer {
|
|
t.Helper()
|
|
for {
|
|
c.mu.Lock()
|
|
if len(c.timers) > index {
|
|
timer := c.timers[index]
|
|
c.mu.Unlock()
|
|
return timer
|
|
}
|
|
c.mu.Unlock()
|
|
select {
|
|
case <-c.created:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatalf("timer %d was not created", index)
|
|
}
|
|
}
|
|
}
|
|
func (c *manualAttemptClock) count() int {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return len(c.timers)
|
|
}
|
|
|
|
type controlledRunCall struct {
|
|
ctx context.Context
|
|
spec runtime.ExecutionSpec
|
|
sink runtime.EventSink
|
|
}
|
|
|
|
type controlledTunnelCall struct {
|
|
ctx context.Context
|
|
req runtime.ProviderTunnelRequest
|
|
sink runtime.ProviderTunnelSink
|
|
}
|
|
|
|
type controlledWatchdogAdapter struct {
|
|
name string
|
|
runCalls chan controlledRunCall
|
|
tunnelCalls chan controlledTunnelCall
|
|
runReturn chan error
|
|
tunnelReturn chan error
|
|
maxConcurrent int
|
|
}
|
|
|
|
func newControlledWatchdogAdapter(name string) *controlledWatchdogAdapter {
|
|
return &controlledWatchdogAdapter{
|
|
name: name, runCalls: make(chan controlledRunCall, 1), tunnelCalls: make(chan controlledTunnelCall, 1),
|
|
runReturn: make(chan error, 1), tunnelReturn: make(chan error, 1), maxConcurrent: 1,
|
|
}
|
|
}
|
|
func (a *controlledWatchdogAdapter) Name() string { return a.name }
|
|
func (a *controlledWatchdogAdapter) Capabilities(context.Context) (runtime.Capabilities, error) {
|
|
return runtime.Capabilities{AdapterName: a.name, Targets: []string{"target"}, MaxConcurrency: a.maxConcurrent}, nil
|
|
}
|
|
func (a *controlledWatchdogAdapter) Execute(ctx context.Context, spec runtime.ExecutionSpec, sink runtime.EventSink) error {
|
|
a.runCalls <- controlledRunCall{ctx: ctx, spec: spec, sink: sink}
|
|
return <-a.runReturn
|
|
}
|
|
func (a *controlledWatchdogAdapter) TunnelProvider(ctx context.Context, req runtime.ProviderTunnelRequest, sink runtime.ProviderTunnelSink) error {
|
|
a.tunnelCalls <- controlledTunnelCall{ctx: ctx, req: req, sink: sink}
|
|
return <-a.tunnelReturn
|
|
}
|
|
|
|
// probeCall records one invocation of the injected health probe so tests can
|
|
// assert the probe received an independent, still-live context after the
|
|
// stalled request was canceled.
|
|
type probeCall struct {
|
|
ctx context.Context
|
|
target string
|
|
}
|
|
|
|
type probeReply struct {
|
|
result runtime.ProviderProbeResult
|
|
err error
|
|
}
|
|
|
|
// probingWatchdogAdapter is a controlledWatchdogAdapter that also implements
|
|
// runtime.ProviderProber. ProbeProvider blocks on a channel so tests drive the
|
|
// independent bounded health probe deterministically and observe the context it
|
|
// received.
|
|
type probingWatchdogAdapter struct {
|
|
*controlledWatchdogAdapter
|
|
probeCalls chan probeCall
|
|
probeReturn chan probeReply
|
|
}
|
|
|
|
func newProbingWatchdogAdapter(name string) *probingWatchdogAdapter {
|
|
return &probingWatchdogAdapter{
|
|
controlledWatchdogAdapter: newControlledWatchdogAdapter(name),
|
|
probeCalls: make(chan probeCall, 1),
|
|
probeReturn: make(chan probeReply, 1),
|
|
}
|
|
}
|
|
|
|
func (a *probingWatchdogAdapter) ProbeProvider(ctx context.Context, target string) (runtime.ProviderProbeResult, error) {
|
|
a.probeCalls <- probeCall{ctx: ctx, target: target}
|
|
reply := <-a.probeReturn
|
|
return reply.result, reply.err
|
|
}
|
|
|
|
type watchdogRouter struct{ adapter runtime.ProviderTunnelAdapter }
|
|
|
|
func (r *watchdogRouter) Resolve(_ context.Context, req runtime.RunRequest) (runtime.ExecutionSpec, error) {
|
|
return runtime.ExecutionSpec{
|
|
RunID: req.RunID, Adapter: r.adapter.Name(), Target: req.Target, SessionID: req.SessionID,
|
|
Background: req.Background, Input: req.Input, TimeoutSec: req.TimeoutSec, Metadata: req.Metadata,
|
|
ResponseStallTimeoutMS: req.ResponseStallTimeoutMS,
|
|
}, nil
|
|
}
|
|
func (r *watchdogRouter) ResolveAdapter(ctx context.Context, req runtime.RunRequest) (runtime.ExecutionSpec, runtime.Provider, error) {
|
|
spec, err := r.Resolve(ctx, req)
|
|
return spec, r.adapter, err
|
|
}
|
|
func (r *watchdogRouter) LookupAdapter(name string) (runtime.Provider, error) {
|
|
if name != r.adapter.Name() {
|
|
return nil, fmt.Errorf("adapter %q not found", name)
|
|
}
|
|
return r.adapter, nil
|
|
}
|
|
func (r *watchdogRouter) GetAdapter(name string) (runtime.Provider, bool) {
|
|
if name == r.adapter.Name() {
|
|
return r.adapter, true
|
|
}
|
|
return nil, false
|
|
}
|
|
|
|
func newWatchdogNode(t *testing.T, adapter runtime.ProviderTunnelAdapter, clock *manualAttemptClock) *Node {
|
|
t.Helper()
|
|
st, err := store.New(":memory:", zap.NewNop())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = st.Close() })
|
|
n := New("node-watchdog", &watchdogRouter{adapter: adapter}, st, 0, io.Discard, zap.NewNop(), nil)
|
|
n.watchdogClock = clock
|
|
return n
|
|
}
|
|
|
|
type watchdogPipe struct {
|
|
edge *toki.TcpClient
|
|
sess *transport.Session
|
|
events chan *iop.RunEvent
|
|
frames chan *iop.ProviderTunnelFrame
|
|
}
|
|
|
|
func newWatchdogPipe(t *testing.T) *watchdogPipe {
|
|
t.Helper()
|
|
edgeConn, nodeConn := net.Pipe()
|
|
edgeParsers := toki.ParserMap{
|
|
toki.TypeNameOf(&iop.RunEvent{}): func(b []byte) (proto.Message, error) {
|
|
m := &iop.RunEvent{}
|
|
return m, proto.Unmarshal(b, m)
|
|
},
|
|
toki.TypeNameOf(&iop.ProviderTunnelFrame{}): func(b []byte) (proto.Message, error) {
|
|
m := &iop.ProviderTunnelFrame{}
|
|
return m, proto.Unmarshal(b, m)
|
|
},
|
|
}
|
|
edge := toki.NewTcpClient(edgeConn, 0, 0, edgeParsers)
|
|
nodeClient := toki.NewTcpClient(nodeConn, 0, 0, toki.ParserMap{})
|
|
pipe := &watchdogPipe{
|
|
edge: edge, sess: transport.ExportNewSession(nodeClient, zap.NewNop(), "node-watchdog", "watchdog"),
|
|
events: make(chan *iop.RunEvent, 8), frames: make(chan *iop.ProviderTunnelFrame, 8),
|
|
}
|
|
toki.AddListenerTyped[*iop.RunEvent](&edge.Communicator, func(event *iop.RunEvent) {
|
|
pipe.events <- proto.Clone(event).(*iop.RunEvent)
|
|
})
|
|
toki.AddListenerTyped[*iop.ProviderTunnelFrame](&edge.Communicator, func(frame *iop.ProviderTunnelFrame) {
|
|
pipe.frames <- proto.Clone(frame).(*iop.ProviderTunnelFrame)
|
|
})
|
|
t.Cleanup(func() { _ = edge.Close(); _ = nodeClient.Close() })
|
|
return pipe
|
|
}
|
|
|
|
func waitContextCanceled(t *testing.T, ctx context.Context) {
|
|
t.Helper()
|
|
select {
|
|
case <-ctx.Done():
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("provider context was not canceled")
|
|
}
|
|
}
|
|
|
|
func waitRunEvent(t *testing.T, events <-chan *iop.RunEvent) *iop.RunEvent {
|
|
t.Helper()
|
|
select {
|
|
case event := <-events:
|
|
return event
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("run event was not emitted")
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func waitTunnelFrame(t *testing.T, frames <-chan *iop.ProviderTunnelFrame) *iop.ProviderTunnelFrame {
|
|
t.Helper()
|
|
select {
|
|
case frame := <-frames:
|
|
return frame
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("tunnel frame was not emitted")
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func requireTimerDurations(t *testing.T, timer *manualAttemptTimer, want ...time.Duration) {
|
|
t.Helper()
|
|
got, _ := timer.snapshot()
|
|
if len(got) != len(want) {
|
|
t.Fatalf("timer durations = %v, want %v", got, want)
|
|
}
|
|
for i := range want {
|
|
if got[i] != want[i] {
|
|
t.Fatalf("timer durations = %v, want %v", got, want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func activeAdapterAttempts(n *Node, adapter string) int {
|
|
n.adapterGatesMu.Lock()
|
|
gate := n.adapterGates[adapter]
|
|
n.adapterGatesMu.Unlock()
|
|
if gate == nil {
|
|
return 0
|
|
}
|
|
return gate.activeCount()
|
|
}
|
|
|
|
func TestAttemptObserverProgressResetsAndFenceIsMonotonic(t *testing.T) {
|
|
clock := newManualAttemptClock()
|
|
observer := newAttemptObserver(clock, time.Second)
|
|
timer := clock.waitTimer(t, 0)
|
|
observer.observe(runtime.DispositionNone)
|
|
requireTimerDurations(t, timer, time.Second)
|
|
observer.observe(runtime.DispositionProgress)
|
|
requireTimerDurations(t, timer, time.Second, time.Second)
|
|
timer.fire()
|
|
expiry, valid := observer.expiryForSignal(<-observer.expired())
|
|
if !valid || !observer.claimFence(expiry) || observer.claimFence(expiry) {
|
|
t.Fatal("fence claim was not monotonic")
|
|
}
|
|
observer.observe(runtime.DispositionProgress)
|
|
requireTimerDurations(t, timer, time.Second, time.Second)
|
|
}
|
|
|
|
func TestAttemptObserverCurrentArmSignalSurvivesImmediateFire(t *testing.T) {
|
|
clock := newManualAttemptClock()
|
|
clock.beforeTimerReturn = func(timer *manualAttemptTimer) { timer.fire() }
|
|
observer := newAttemptObserver(clock, time.Nanosecond)
|
|
|
|
expiry, valid := observer.expiryForSignal(<-observer.expired())
|
|
if !valid {
|
|
t.Fatal("current timer signal was rejected because expiry bookkeeping followed the fire")
|
|
}
|
|
if !observer.claimFence(expiry) {
|
|
t.Fatal("current timer signal did not claim the fence")
|
|
}
|
|
}
|
|
|
|
type recordingProtoSender struct {
|
|
mu sync.Mutex
|
|
messages []proto.Message
|
|
entered chan struct{}
|
|
release chan struct{}
|
|
once sync.Once
|
|
}
|
|
|
|
func (s *recordingProtoSender) Send(message proto.Message) error {
|
|
if s.entered != nil {
|
|
s.once.Do(func() {
|
|
close(s.entered)
|
|
<-s.release
|
|
})
|
|
}
|
|
s.mu.Lock()
|
|
s.messages = append(s.messages, proto.Clone(message))
|
|
s.mu.Unlock()
|
|
return nil
|
|
}
|
|
func (s *recordingProtoSender) snapshot() []proto.Message {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return append([]proto.Message(nil), s.messages...)
|
|
}
|
|
|
|
func TestTunnelSinkStallClaimSerializesAcceptedFrame(t *testing.T) {
|
|
clock := newManualAttemptClock()
|
|
sender := &recordingProtoSender{entered: make(chan struct{}), release: make(chan struct{})}
|
|
sink := &tunnelSink{sess: sender, observer: newAttemptObserver(clock, time.Second)}
|
|
bodyDone := make(chan error, 1)
|
|
go func() {
|
|
bodyDone <- sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{Kind: runtime.ProviderTunnelFrameKindBody, Body: []byte("accepted")})
|
|
}()
|
|
<-sender.entered
|
|
if sink.mu.TryLock() {
|
|
sink.mu.Unlock()
|
|
t.Fatal("tunnel emission lock was released before accepted frame Send completed")
|
|
}
|
|
close(sender.release)
|
|
if err := <-bodyDone; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
timer := clock.waitTimer(t, 0)
|
|
timer.fire()
|
|
expiry, valid := sink.observer.expiryForSignal(<-sink.observer.expired())
|
|
if !valid || !sink.claimStall(expiry) {
|
|
t.Fatal("stall claim failed after accepted frame completed")
|
|
}
|
|
if err := sink.emitClaimedTerminal(context.Background(), stalledTunnelFrame(runtime.ProviderTunnelRequest{RunID: "run", TunnelID: "tunnel"}, stallObservation{fence: "confirmed", idle: time.Second})); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{Kind: runtime.ProviderTunnelFrameKindUsage}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
messages := sender.snapshot()
|
|
if len(messages) != 2 {
|
|
t.Fatalf("sent frames = %d, want body then terminal", len(messages))
|
|
}
|
|
if messages[0].(*iop.ProviderTunnelFrame).GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY || messages[1].(*iop.ProviderTunnelFrame).GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR {
|
|
t.Fatalf("frame order = %v, %v", messages[0], messages[1])
|
|
}
|
|
}
|
|
|
|
func TestRunWatchdogStaleExpiryYieldsToProgress(t *testing.T) {
|
|
clock := newManualAttemptClock()
|
|
adapter := newControlledWatchdogAdapter("run-stale-expiry")
|
|
n := newWatchdogNode(t, adapter, clock)
|
|
pipe := newWatchdogPipe(t)
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{RunId: "run-stale-expiry", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
|
|
}()
|
|
call := <-adapter.runCalls
|
|
sink := call.sink.(*terminalDeferringSink)
|
|
claimStarted := make(chan struct{})
|
|
releaseClaim := make(chan struct{})
|
|
claimResult := make(chan bool, 1)
|
|
var firstClaim sync.Once
|
|
sink.beforeStallClaim = func() {
|
|
firstClaim.Do(func() {
|
|
close(claimStarted)
|
|
<-releaseClaim
|
|
})
|
|
}
|
|
sink.afterStallClaim = func(claimed bool) { claimResult <- claimed }
|
|
|
|
stallTimer := clock.waitTimer(t, 0)
|
|
stallTimer.fire()
|
|
<-claimStarted // The old timer was consumed before provider progress arrives.
|
|
if err := call.sink.Emit(context.Background(), runtime.RuntimeEvent{RunID: "run-stale-expiry", Type: runtime.EventTypeDelta, Delta: "progress"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if event := waitRunEvent(t, pipe.events); event.GetType() != string(runtime.EventTypeDelta) {
|
|
t.Fatalf("progress event = %+v", event)
|
|
}
|
|
requireTimerDurations(t, stallTimer, time.Second, time.Second)
|
|
close(releaseClaim)
|
|
if claimed := <-claimResult; claimed {
|
|
t.Fatal("stale normalized expiry fenced after progress reset the watchdog")
|
|
}
|
|
select {
|
|
case event := <-pipe.events:
|
|
t.Fatalf("stale normalized expiry emitted terminal: %+v", event)
|
|
default:
|
|
}
|
|
|
|
stallTimer.fire()
|
|
waitContextCanceled(t, call.ctx)
|
|
grace := clock.waitTimer(t, 1)
|
|
adapter.runReturn <- nil
|
|
if err := <-done; err != errProviderResponseStalled {
|
|
t.Fatalf("run result = %v", err)
|
|
}
|
|
if terminal := waitRunEvent(t, pipe.events); terminal.GetType() != string(runtime.EventTypeError) || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
|
|
t.Fatalf("terminal = %+v", terminal)
|
|
}
|
|
if claimed := <-claimResult; !claimed {
|
|
t.Fatal("reset normalized expiry did not claim the watchdog fence")
|
|
}
|
|
requireTimerDurations(t, grace, defaultAttemptCloseGrace)
|
|
}
|
|
|
|
func TestTunnelWatchdogStaleExpiryYieldsToProgress(t *testing.T) {
|
|
clock := newManualAttemptClock()
|
|
adapter := newControlledWatchdogAdapter("tunnel-stale-expiry")
|
|
n := newWatchdogNode(t, adapter, clock)
|
|
pipe := newWatchdogPipe(t)
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{RunId: "tunnel-stale-expiry", TunnelId: "tunnel", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
|
|
}()
|
|
call := <-adapter.tunnelCalls
|
|
sink := call.sink.(*tunnelSink)
|
|
claimStarted := make(chan struct{})
|
|
releaseClaim := make(chan struct{})
|
|
claimResult := make(chan bool, 1)
|
|
var firstClaim sync.Once
|
|
sink.beforeStallClaim = func() {
|
|
firstClaim.Do(func() {
|
|
close(claimStarted)
|
|
<-releaseClaim
|
|
})
|
|
}
|
|
sink.afterStallClaim = func(claimed bool) { claimResult <- claimed }
|
|
|
|
stallTimer := clock.waitTimer(t, 0)
|
|
stallTimer.fire()
|
|
<-claimStarted // The old timer was consumed before the accepted frame progresses the attempt.
|
|
if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{RunID: "tunnel-stale-expiry", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindBody, Body: []byte("progress")}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if frame := waitTunnelFrame(t, pipe.frames); frame.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY {
|
|
t.Fatalf("progress frame = %+v", frame)
|
|
}
|
|
requireTimerDurations(t, stallTimer, time.Second, time.Second)
|
|
close(releaseClaim)
|
|
if claimed := <-claimResult; claimed {
|
|
t.Fatal("stale tunnel expiry fenced after progress reset the watchdog")
|
|
}
|
|
select {
|
|
case frame := <-pipe.frames:
|
|
t.Fatalf("stale tunnel expiry emitted terminal: %+v", frame)
|
|
default:
|
|
}
|
|
|
|
stallTimer.fire()
|
|
waitContextCanceled(t, call.ctx)
|
|
grace := clock.waitTimer(t, 1)
|
|
adapter.tunnelReturn <- nil
|
|
if err := <-done; err != errProviderResponseStalled {
|
|
t.Fatalf("tunnel result = %v", err)
|
|
}
|
|
if terminal := waitTunnelFrame(t, pipe.frames); terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
|
|
t.Fatalf("terminal = %+v", terminal)
|
|
}
|
|
if claimed := <-claimResult; !claimed {
|
|
t.Fatal("reset tunnel expiry did not claim the watchdog fence")
|
|
}
|
|
requireTimerDurations(t, grace, defaultAttemptCloseGrace)
|
|
}
|
|
|
|
func TestRunWatchdogOldArmFireDuringResetYieldsToProgress(t *testing.T) {
|
|
clock := newManualAttemptClock()
|
|
adapter := newControlledWatchdogAdapter("run-old-arm-during-reset")
|
|
n := newWatchdogNode(t, adapter, clock)
|
|
pipe := newWatchdogPipe(t)
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{RunId: "run-old-arm-during-reset", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
|
|
}()
|
|
call := <-adapter.runCalls
|
|
sink := call.sink.(*terminalDeferringSink)
|
|
captureResults := make(chan bool, 2)
|
|
sink.observer.afterExpiryCapture = func(valid bool) { captureResults <- valid }
|
|
|
|
stallTimer := clock.waitTimer(t, 0)
|
|
clock.advanceTo(clock.current().Add(time.Second))
|
|
stallTimer.beforeReset = stallTimer.fireStaleArmDuringReset
|
|
if err := call.sink.Emit(context.Background(), runtime.RuntimeEvent{RunID: "run-old-arm-during-reset", Type: runtime.EventTypeDelta, Delta: "progress"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if event := waitRunEvent(t, pipe.events); event.GetType() != string(runtime.EventTypeDelta) {
|
|
t.Fatalf("progress event = %+v", event)
|
|
}
|
|
if valid := <-captureResults; valid {
|
|
t.Fatal("old normalized arm was accepted while progress reset the watchdog")
|
|
}
|
|
if err := call.ctx.Err(); err != nil {
|
|
t.Fatal("old normalized arm canceled the provider before the reset threshold")
|
|
}
|
|
requireTimerDurations(t, stallTimer, time.Second, time.Second)
|
|
|
|
stallTimer.fire()
|
|
waitContextCanceled(t, call.ctx)
|
|
grace := clock.waitTimer(t, 1)
|
|
adapter.runReturn <- nil
|
|
if err := <-done; err != errProviderResponseStalled {
|
|
t.Fatalf("run result = %v", err)
|
|
}
|
|
if valid := <-captureResults; !valid {
|
|
t.Fatal("reset normalized arm was not accepted after its full threshold")
|
|
}
|
|
if terminal := waitRunEvent(t, pipe.events); terminal.GetType() != string(runtime.EventTypeError) || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
|
|
t.Fatalf("terminal = %+v", terminal)
|
|
}
|
|
requireTimerDurations(t, grace, defaultAttemptCloseGrace)
|
|
}
|
|
|
|
func TestTunnelWatchdogOldArmFireDuringResetYieldsToProgress(t *testing.T) {
|
|
clock := newManualAttemptClock()
|
|
adapter := newControlledWatchdogAdapter("tunnel-old-arm-during-reset")
|
|
n := newWatchdogNode(t, adapter, clock)
|
|
pipe := newWatchdogPipe(t)
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{RunId: "tunnel-old-arm-during-reset", TunnelId: "tunnel", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
|
|
}()
|
|
call := <-adapter.tunnelCalls
|
|
sink := call.sink.(*tunnelSink)
|
|
captureResults := make(chan bool, 2)
|
|
sink.observer.afterExpiryCapture = func(valid bool) { captureResults <- valid }
|
|
|
|
stallTimer := clock.waitTimer(t, 0)
|
|
clock.advanceTo(clock.current().Add(time.Second))
|
|
stallTimer.beforeReset = stallTimer.fireStaleArmDuringReset
|
|
if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{RunID: "tunnel-old-arm-during-reset", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindBody, Body: []byte("progress")}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if frame := waitTunnelFrame(t, pipe.frames); frame.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY {
|
|
t.Fatalf("progress frame = %+v", frame)
|
|
}
|
|
if valid := <-captureResults; valid {
|
|
t.Fatal("old tunnel arm was accepted while progress reset the watchdog")
|
|
}
|
|
if err := call.ctx.Err(); err != nil {
|
|
t.Fatal("old tunnel arm canceled the provider before the reset threshold")
|
|
}
|
|
requireTimerDurations(t, stallTimer, time.Second, time.Second)
|
|
|
|
stallTimer.fire()
|
|
waitContextCanceled(t, call.ctx)
|
|
grace := clock.waitTimer(t, 1)
|
|
adapter.tunnelReturn <- nil
|
|
if err := <-done; err != errProviderResponseStalled {
|
|
t.Fatalf("tunnel result = %v", err)
|
|
}
|
|
if valid := <-captureResults; !valid {
|
|
t.Fatal("reset tunnel arm was not accepted after its full threshold")
|
|
}
|
|
if terminal := waitTunnelFrame(t, pipe.frames); terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
|
|
t.Fatalf("terminal = %+v", terminal)
|
|
}
|
|
requireTimerDurations(t, grace, defaultAttemptCloseGrace)
|
|
}
|
|
|
|
func TestRunWatchdogStaleExpiryBeforeCaptureYieldsToProgress(t *testing.T) {
|
|
clock := newManualAttemptClock()
|
|
adapter := newControlledWatchdogAdapter("run-stale-before-capture")
|
|
n := newWatchdogNode(t, adapter, clock)
|
|
pipe := newWatchdogPipe(t)
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- n.OnRunRequest(context.Background(), pipe.sess, &iop.RunRequest{RunId: "run-stale-before-capture", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
|
|
}()
|
|
call := <-adapter.runCalls
|
|
sink := call.sink.(*terminalDeferringSink)
|
|
captureStarted := make(chan struct{})
|
|
releaseCapture := make(chan struct{})
|
|
captureResults := make(chan bool, 4)
|
|
var firstCapture sync.Once
|
|
sink.observer.beforeExpiryCapture = func() {
|
|
firstCapture.Do(func() {
|
|
close(captureStarted)
|
|
<-releaseCapture
|
|
})
|
|
}
|
|
sink.observer.afterExpiryCapture = func(valid bool) { captureResults <- valid }
|
|
|
|
stallTimer := clock.waitTimer(t, 0)
|
|
stallTimer.fire()
|
|
<-captureStarted // The old timer signal was received before its validity is captured.
|
|
if err := call.sink.Emit(context.Background(), runtime.RuntimeEvent{RunID: "run-stale-before-capture", Type: runtime.EventTypeDelta, Delta: "progress"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if event := waitRunEvent(t, pipe.events); event.GetType() != string(runtime.EventTypeDelta) {
|
|
t.Fatalf("progress event = %+v", event)
|
|
}
|
|
requireTimerDurations(t, stallTimer, time.Second, time.Second)
|
|
close(releaseCapture)
|
|
if valid := <-captureResults; valid {
|
|
t.Fatal("stale normalized expiry captured as valid after progress reset the watchdog")
|
|
}
|
|
if err := call.ctx.Err(); err != nil {
|
|
t.Fatal("stale normalized expiry canceled the provider before its reset threshold")
|
|
}
|
|
select {
|
|
case event := <-pipe.events:
|
|
t.Fatalf("stale normalized expiry emitted terminal: %+v", event)
|
|
default:
|
|
}
|
|
|
|
stallTimer.fire()
|
|
waitContextCanceled(t, call.ctx)
|
|
grace := clock.waitTimer(t, 1)
|
|
adapter.runReturn <- nil
|
|
if err := <-done; err != errProviderResponseStalled {
|
|
t.Fatalf("run result = %v", err)
|
|
}
|
|
if valid := <-captureResults; !valid {
|
|
t.Fatal("reset normalized expiry was not captured as valid after its full threshold")
|
|
}
|
|
if terminal := waitRunEvent(t, pipe.events); terminal.GetType() != string(runtime.EventTypeError) || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
|
|
t.Fatalf("terminal = %+v", terminal)
|
|
}
|
|
requireTimerDurations(t, grace, defaultAttemptCloseGrace)
|
|
}
|
|
|
|
func TestTunnelWatchdogStaleExpiryBeforeCaptureYieldsToProgress(t *testing.T) {
|
|
clock := newManualAttemptClock()
|
|
adapter := newControlledWatchdogAdapter("tunnel-stale-before-capture")
|
|
n := newWatchdogNode(t, adapter, clock)
|
|
pipe := newWatchdogPipe(t)
|
|
done := make(chan error, 1)
|
|
go func() {
|
|
done <- n.OnProviderTunnelRequest(context.Background(), pipe.sess, &iop.ProviderTunnelRequest{RunId: "tunnel-stale-before-capture", TunnelId: "tunnel", Adapter: adapter.Name(), Target: "target", ResponseStallTimeoutMs: 1000})
|
|
}()
|
|
call := <-adapter.tunnelCalls
|
|
sink := call.sink.(*tunnelSink)
|
|
captureStarted := make(chan struct{})
|
|
releaseCapture := make(chan struct{})
|
|
captureResults := make(chan bool, 4)
|
|
var firstCapture sync.Once
|
|
sink.observer.beforeExpiryCapture = func() {
|
|
firstCapture.Do(func() {
|
|
close(captureStarted)
|
|
<-releaseCapture
|
|
})
|
|
}
|
|
sink.observer.afterExpiryCapture = func(valid bool) { captureResults <- valid }
|
|
|
|
stallTimer := clock.waitTimer(t, 0)
|
|
stallTimer.fire()
|
|
<-captureStarted // The old timer signal was received before its validity is captured.
|
|
if err := call.sink.EmitTunnelFrame(context.Background(), runtime.ProviderTunnelFrame{RunID: "tunnel-stale-before-capture", TunnelID: "tunnel", Kind: runtime.ProviderTunnelFrameKindBody, Body: []byte("progress")}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if frame := waitTunnelFrame(t, pipe.frames); frame.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_BODY {
|
|
t.Fatalf("progress frame = %+v", frame)
|
|
}
|
|
requireTimerDurations(t, stallTimer, time.Second, time.Second)
|
|
close(releaseCapture)
|
|
if valid := <-captureResults; valid {
|
|
t.Fatal("stale tunnel expiry captured as valid after progress reset the watchdog")
|
|
}
|
|
if err := call.ctx.Err(); err != nil {
|
|
t.Fatal("stale tunnel expiry canceled the provider before its reset threshold")
|
|
}
|
|
select {
|
|
case frame := <-pipe.frames:
|
|
t.Fatalf("stale tunnel expiry emitted terminal: %+v", frame)
|
|
default:
|
|
}
|
|
|
|
stallTimer.fire()
|
|
waitContextCanceled(t, call.ctx)
|
|
grace := clock.waitTimer(t, 1)
|
|
adapter.tunnelReturn <- nil
|
|
if err := <-done; err != errProviderResponseStalled {
|
|
t.Fatalf("tunnel result = %v", err)
|
|
}
|
|
if valid := <-captureResults; !valid {
|
|
t.Fatal("reset tunnel expiry was not captured as valid after its full threshold")
|
|
}
|
|
if terminal := waitTunnelFrame(t, pipe.frames); terminal.GetKind() != iop.ProviderTunnelFrameKind_PROVIDER_TUNNEL_FRAME_KIND_ERROR || terminal.GetMetadata()["attempt_fence"] != "confirmed" {
|
|
t.Fatalf("terminal = %+v", terminal)
|
|
}
|
|
requireTimerDurations(t, grace, defaultAttemptCloseGrace)
|
|
}
|