Answer an ask the build module cannot read on its step key as refused (issue 1921, seams must 1)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -561,6 +561,36 @@ func (p *Puller) declineRun(msg *Pulled, ask delivery.StepAsk, key string, step
|
||||
p.mu.Unlock()
|
||||
}
|
||||
|
||||
// refuseUnread ends an ask this module cannot read: when its data holds an id that keys a step, the step is written
|
||||
// refused with the reason, so the asker learns at once that no node began it; otherwise (no id, not JSON) the ask is only
|
||||
// dropped and logged. The message is dropped either way.
|
||||
func (p *Puller) refuseUnread(msg *Pulled, why error) {
|
||||
var probe struct {
|
||||
ID string `json:"id"`
|
||||
Kind string `json:"kind"`
|
||||
}
|
||||
if json.Unmarshal(msg.Data, &probe) == nil && probe.ID != "" {
|
||||
if key, kerr := delivery.AskStepKey(p.Node, probe.ID); kerr == nil {
|
||||
tool := "ask"
|
||||
if probe.Kind != "" {
|
||||
tool = probe.Kind + " ask"
|
||||
}
|
||||
at := now().UTC()
|
||||
step := delivery.Step{ID: probe.ID, Tool: tool, Node: p.Node, State: delivery.StepRefused, Started: at, Ended: at,
|
||||
Reason: fmt.Sprintf("this build module on %s cannot read the ask: %v", p.Node, why), Owner: sdkstep.Owner}
|
||||
if werr := writeEnd(p.Put, key, step); werr != nil {
|
||||
p.logf("the unread ask %s cannot be answered on %s: %v", probe.ID, p.Node, werr)
|
||||
} else {
|
||||
p.logf("an ask refused: %v", why)
|
||||
_ = msg.Term()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
p.logf("an ask dropped: %v", why)
|
||||
_ = msg.Term()
|
||||
}
|
||||
|
||||
func withKind(msg *Pulled, kind string) *Pulled {
|
||||
if msg != nil {
|
||||
msg.Kind = kind
|
||||
@@ -576,9 +606,10 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
|
||||
err = fmt.Errorf("the ask %s is a %s ask of %s, pulled from the %s %s worker", ask.ID, ask.Kind, ask.Language, p.Language, kind)
|
||||
}
|
||||
if err != nil {
|
||||
// An ask that is not one runs nowhere: delivering it again would only fail again.
|
||||
p.logf("an ask dropped: %v", err)
|
||||
_ = msg.Term()
|
||||
// An ask that is not one runs nowhere: delivering it again would only fail again. But an ask that names an id
|
||||
// is answered on its step key as refused, naming why (a job this build module does not know, a version it does
|
||||
// not read): a dropped ask must always be answered, or its asker waits to its bound for nothing.
|
||||
p.refuseUnread(msg, err)
|
||||
return
|
||||
}
|
||||
if lacks := delivery.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
|
||||
|
||||
@@ -519,3 +519,36 @@ func TestAPoisonedAskIsSaidAndRunNoMore(t *testing.T) {
|
||||
t.Errorf("acked %v naked %v, want it termed alone", acked, naked)
|
||||
}
|
||||
}
|
||||
|
||||
// An ask whose job this build module does not know is answered on its step key as refused, naming why, never only dropped
|
||||
// (a dropped ask leaves its asker waiting to its bound); one with no id is still dropped.
|
||||
func TestPullAnswersAnAskItCannotRead(t *testing.T) {
|
||||
src, st := NewFakeAsks(), newSteps()
|
||||
good := testAsk(delivery.AskCheck)
|
||||
raw, err := delivery.EncodeStepAsk(good)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m map[string]any
|
||||
_ = json.Unmarshal(raw, &m)
|
||||
delete(m, "check")
|
||||
m["frobnicate"] = map[string]any{"program": "x"}
|
||||
odd, _ := json.Marshal(m)
|
||||
// The job is not one this reader holds, so the ask carries none it knows: decode refuses it.
|
||||
if _, err := delivery.DecodeStepAsk(odd); err == nil {
|
||||
t.Fatal("the fixture ask decodes")
|
||||
}
|
||||
src.Publish(delivery.AskCheck, odd)
|
||||
src.Publish(delivery.AskCheck, []byte(`{"v":1}`))
|
||||
p := &Puller{Source: src, Language: delivery.AskGo, Node: "ace", Put: st.put, Capacity: func() int { return 1 },
|
||||
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { t.Error("an unread ask ran"); return nil, nil }}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
go p.Pull(ctx)
|
||||
waitFor(t, "both termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 2 })
|
||||
key, _ := delivery.AskStepKey("ace", good.ID)
|
||||
s, ok := st.get(key)
|
||||
if !ok || s.State != delivery.StepRefused || !strings.Contains(s.Reason, "cannot read the ask") || !strings.Contains(s.Reason, "exactly one") {
|
||||
t.Fatalf("the unread ask's step = %+v (%v)", s, ok)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user