diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index f5ea8c3..0f8540a 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -2,6 +2,7 @@ package buildrun import ( "context" + "encoding/json" "errors" "strings" "sync" @@ -9,6 +10,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" ) func goAsk(id string) delivery.StepAsk { @@ -106,6 +108,18 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if err != nil || !IsLost(step) { t.Fatalf("an ask heard given up after the hearing came back: %+v, %v", step, err) } + for _, row := range []struct { + h *Hearing + want string + }{{nil, health.StateUnhealthy}, {&told[0], health.StateUnhealthy}, {&told[1], health.StateHealthy}} { + var raw []byte + if row.h != nil { + raw, _ = json.Marshal(row.h) + } + if a := HearingHealth(raw); a.State != row.want || a.Why == "" { + t.Errorf("health of %+v = %+v, want %s", row.h, a, row.want) + } + } } // Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the diff --git a/go/delivery/buildrun/lostask.go b/go/delivery/buildrun/lostask.go index 1e70245..90a9ece 100644 --- a/go/delivery/buildrun/lostask.go +++ b/go/delivery/buildrun/lostask.go @@ -2,6 +2,7 @@ package buildrun import ( "context" + "encoding/json" "fmt" "strconv" "strings" @@ -11,6 +12,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" ) // **An ask the build seat gave up on ends its step** (novox/hq issue 734). A build seat's worker offers an ask @@ -451,3 +453,43 @@ func (h *DeadLetterHearing) tell() { h.said(now) } } + +// HearingState is the name of the state an asker keeps its Hearing in (`_hearing`, key HearingKey), which its +// health tool reads. +const HearingState = "hearing" + +// SayHearing is a DeadLetterHearing's said that keeps each Hearing in kv under HearingKey; nil kv keeps nothing and +// says so once. +func SayHearing(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Hearing) { + if kv == nil { + if logf != nil { + logf("whether %s is heard is not kept: the state %s is not there, so the health tool cannot say it", DeadLettersStream, HearingState) + } + return nil + } + return func(h Hearing) { + data, err := json.Marshal(h) + if err == nil { + _, err = kv.Put(ctx, HearingKey, data) + } + if err != nil && logf != nil { + logf("whether %s is heard could not be kept: %v", DeadLettersStream, err) + } + } +} + +// HearingHealth is the health answer of an asker's kept Hearing (raw, empty when none is kept): healthy while it hears +// DEAD_LETTERS, unhealthy saying why while it does not or has not said. +func HearingHealth(raw []byte) health.Answer { + if len(raw) == 0 { + return health.Unhealthy("it has not said whether it hears " + DeadLettersStream + " (issue 734): it is starting, or its state " + HearingState + " is not there") + } + var h Hearing + if err := json.Unmarshal(raw, &h); err != nil { + return health.Unhealthy("what it keeps of hearing " + DeadLettersStream + " is not a hearing: " + err.Error()) + } + if !h.Heard { + return health.Unhealthy(h.Why + " (since " + h.At.Format(time.RFC3339) + "; tried again at its next reconnect or ask)") + } + return health.Healthy("it hears " + DeadLettersStream + " for " + strings.Join(h.Streams, ", ")) +}