diff --git a/modules/mesh-build-go/buildstep/pull.go b/modules/mesh-build-go/buildstep/pull.go index b7e54ac6d..9f5ed04b4 100644 --- a/modules/mesh-build-go/buildstep/pull.go +++ b/modules/mesh-build-go/buildstep/pull.go @@ -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 { diff --git a/modules/mesh-build-go/buildstep/pull_test.go b/modules/mesh-build-go/buildstep/pull_test.go index 086b15527..d196d8eec 100644 --- a/modules/mesh-build-go/buildstep/pull_test.go +++ b/modules/mesh-build-go/buildstep/pull_test.go @@ -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) + } +}