135 lines
4.6 KiB
Go
135 lines
4.6 KiB
Go
package buildstep
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
// 1887/A11: a bundle's source pulls the kind's worker by its seat's name and settles each ask by the runtime's id.
|
|
func TestSeatAsksPullsTheKindsWorkerAndSettlesByID(t *testing.T) {
|
|
var pulled []string
|
|
var settled []string
|
|
src := SeatAsks{Language: sdkstep.AskGo,
|
|
Pull: func(worker string, wait time.Duration) (*stdio.Pulled, error) {
|
|
pulled = append(pulled, worker)
|
|
if worker != "SEAT_NODE_BUILD_GO_CHECK_worker" {
|
|
return nil, nil
|
|
}
|
|
return &stdio.Pulled{ID: "p9", Worker: worker, Body: []byte(`{"v":1}`), Delivered: 2}, nil
|
|
},
|
|
Settle: func(id, fate string) error { settled = append(settled, id+" "+fate); return nil }}
|
|
if p, err := src.Next(context.Background(), sdkstep.AskDelivery, 0); err != nil || p != nil {
|
|
t.Fatalf("delivery = %+v, %v", p, err)
|
|
}
|
|
p, err := src.Next(context.Background(), sdkstep.AskCheck, time.Second)
|
|
if err != nil || p == nil || string(p.Data) != `{"v":1}` || p.Delivered != 2 {
|
|
t.Fatalf("check = %+v, %v", p, err)
|
|
}
|
|
_ = p.InProgress()
|
|
_ = p.Ack()
|
|
if len(pulled) != 2 || pulled[0] != "SEAT_NODE_BUILD_GO_DELIVERY_worker" {
|
|
t.Errorf("pulled %v", pulled)
|
|
}
|
|
if len(settled) != 2 || settled[0] != "p9 progress" || settled[1] != "p9 ack" {
|
|
t.Errorf("settled %v", settled)
|
|
}
|
|
}
|
|
|
|
// pauseKV is a pauses state in memory: its put, load, delete and keys.
|
|
type pauseKV struct {
|
|
mu sync.Mutex
|
|
m map[string][]byte
|
|
}
|
|
|
|
func (k *pauseKV) put(key string, v any) (uint64, error) {
|
|
raw, _ := v.(json.RawMessage)
|
|
k.mu.Lock()
|
|
defer k.mu.Unlock()
|
|
if k.m == nil {
|
|
k.m = map[string][]byte{}
|
|
}
|
|
k.m[key] = raw
|
|
return 1, nil
|
|
}
|
|
func (k *pauseKV) load(key string) ([]byte, error) {
|
|
k.mu.Lock()
|
|
defer k.mu.Unlock()
|
|
return k.m[key], nil
|
|
}
|
|
func (k *pauseKV) forget(key string) error {
|
|
k.mu.Lock()
|
|
defer k.mu.Unlock()
|
|
delete(k.m, key)
|
|
return nil
|
|
}
|
|
func (k *pauseKV) keys() ([]string, error) {
|
|
k.mu.Lock()
|
|
defer k.mu.Unlock()
|
|
var out []string
|
|
for key := range k.m {
|
|
out = append(out, key)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// ADR 0664 §5 (issue 1748): the seat's verb drop. An ask dropped before it is given is refused at pull, naming who dropped it
|
|
// and why, its key deleted, and never run; one that runs when it is dropped has its step's context cancelled and ends
|
|
// refused the same way, acknowledged; the start deletes drops older than DroppedKeep.
|
|
func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) {
|
|
for _, running := range []bool{false, true} {
|
|
src, st, kv := NewFakeAsks(), newSteps(), &pauseKV{}
|
|
a := testAsk(sdkstep.AskDelivery)
|
|
started := make(chan struct{})
|
|
ran := false
|
|
p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, PausePut: kv.put, Load: kv.load, ForgetPause: kv.forget, PauseKeys: kv.keys,
|
|
Capacity: func() int { return 1 }, PullWait: 10 * time.Millisecond,
|
|
Run: func(ctx context.Context, _ sdkstep.StepAsk) (any, error) {
|
|
ran = true
|
|
close(started)
|
|
<-ctx.Done()
|
|
return nil, ctx.Err()
|
|
}}
|
|
if !running {
|
|
if err := p.Drop(a.ID, "mesh-delivery-builder", "superseded"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
_ = src.PublishAsk(a)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
go p.Pull(ctx)
|
|
if running {
|
|
<-started
|
|
if err := p.Drop(a.ID, "mesh-delivery-builder", "superseded"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
waitFor(t, "the ask ended", func() bool { ack, tm, _ := src.Fates(); return len(ack)+len(tm) == 1 })
|
|
cancel()
|
|
key, _ := sdkstep.AskStepKey("shanks", a.ID)
|
|
s, ok := st.get(key)
|
|
if !ok || s.State != sdkstep.StepRefused || !strings.Contains(s.Reason, "dropped by mesh-delivery-builder: superseded") || ran != running {
|
|
t.Errorf("running %v: step %+v (%v), ran %v", running, s, ok, ran)
|
|
}
|
|
waitFor(t, fmt.Sprint("the drop key deleted, running ", running), func() bool { kv.mu.Lock(); defer kv.mu.Unlock(); _, held := kv.m[sdkstep.DropKey(a.ID)]; return !held })
|
|
}
|
|
// The start deletes a drop older than DroppedKeep and keeps a newer one.
|
|
kv := &pauseKV{}
|
|
old, _ := json.Marshal(sdkstep.Drop{V: sdkstep.Version, Why: "old", At: time.Now().Add(-DroppedKeep - time.Hour)})
|
|
fresh, _ := json.Marshal(sdkstep.Drop{V: sdkstep.Version, Why: "fresh", At: time.Now()})
|
|
kv.put(sdkstep.DropKey("old"), json.RawMessage(old))
|
|
kv.put(sdkstep.DropKey("fresh"), json.RawMessage(fresh))
|
|
p := &Puller{Load: kv.load, ForgetPause: kv.forget, PauseKeys: kv.keys}
|
|
p.sweepDropped()
|
|
if keys, _ := kv.keys(); len(keys) != 1 || keys[0] != sdkstep.DropKey("fresh") {
|
|
t.Errorf("keys after the sweep = %v, want the fresh drop alone", keys)
|
|
}
|
|
}
|