G-delivery 1: the builder's bounds not watched, or a bound not written, is a finding
A watch of bounds that cannot start lost every bound's end with a log line only: it is now family-not-heard naming the state. A publish bound that cannot be written would never end: the version is publish-unanswered at once, saying why. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -381,11 +381,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
pubs.ended(ctx, e)
|
||||
}
|
||||
if pw, err := settle(ctx, kv, each); err != nil {
|
||||
process.NotWatched(logger.Printf, versions.BoundsBucket, err, "a publish no holder answers, and a node holding the old form seven days, are not raised until the builder starts again")
|
||||
} else {
|
||||
go func() { errs <- follow(ctx, pw, each) }()
|
||||
}
|
||||
watchBounds(ctx, kv, each, found.notHeard(ctx), errs, logger.Printf)
|
||||
})
|
||||
boundsKV.Ready(func(kv jetstream.KeyValue) {
|
||||
each := func(e jetstream.KeyValueEntry) { bounds.ended(ctx, e) }
|
||||
|
||||
@@ -9,6 +9,9 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/sdk/heard"
|
||||
|
||||
"mesh-delivery-builder/internal/versions"
|
||||
)
|
||||
@@ -24,16 +27,31 @@ type publishes struct {
|
||||
logf func(string, ...any)
|
||||
}
|
||||
|
||||
// arm writes the version's bound when none stands, so a standing bound keeps its time.
|
||||
// arm writes the version's bound when none stands, so a standing bound keeps its time. A bound that cannot be written
|
||||
// would never end, so the version is publish-unanswered at once (review round 2): the finding says why.
|
||||
func (p *publishes) arm(ctx context.Context, e versions.Entry) {
|
||||
kv := p.bounds()
|
||||
var err error
|
||||
if kv == nil {
|
||||
p.logf("the publish bound of %s@%s is not written (%s is not on the bus yet), so it is not raised if no holder answers", e.Name, e.Version, versions.BoundsBucket)
|
||||
err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket)
|
||||
} else if _, err = kv.Create(ctx, versions.PublishBoundKey(e.Name, e.Version), versions.EncodeBound("the publish of "+e.Name+"@"+e.Version+" to "+e.Seat, time.Now().Add(p.bound)), jetstream.KeyTTL(p.bound)); errors.Is(err, jetstream.ErrKeyExists) {
|
||||
err = nil
|
||||
}
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
if _, err := kv.Create(ctx, versions.PublishBoundKey(e.Name, e.Version), versions.EncodeBound("the publish of "+e.Name+"@"+e.Version+" to "+e.Seat, time.Now().Add(p.bound)), jetstream.KeyTTL(p.bound)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) {
|
||||
p.logf("the publish bound of %s@%s is not written, so it is not raised if no holder answers: %v", e.Name, e.Version, err)
|
||||
}
|
||||
p.logf("the publish bound of %s@%s is not written, so it is raised unanswered now: %v", e.Name, e.Version, err)
|
||||
p.unanswered(ctx, e, fmt.Sprintf("its bound could not be written (%v), so the builder cannot wait on the holder's word", err))
|
||||
}
|
||||
|
||||
// unanswered raises publish-unanswered about e's name, saying why.
|
||||
func (p *publishes) unanswered(ctx context.Context, v versions.Entry, why string) {
|
||||
p.findings.raise(ctx, finding{kind: versions.PublishUnanswered, scope: health.ScopeModule, id: v.Name,
|
||||
summary: fmt.Sprintf("%s said neither published nor refused of %s@%s: %s", v.Seat, v.Name, v.Version, why),
|
||||
headline: "A registry seat did not answer a publish",
|
||||
explanation: "The builder built and numbered " + v.Name + " " + v.Version + " and asked " + v.Seat + " to publish it; no word of its holder is known, so the build says done while the version may not be in the registry.",
|
||||
resolved: "That version is published, or a later version of " + v.Name + " is.",
|
||||
needs: "a look at the holder of " + v.Seat + " (its health and log); builder_publish_again asks the publish again."})
|
||||
}
|
||||
|
||||
// heard is told each holder's word the numberer took in: published clears the name's findings, refused raises
|
||||
@@ -93,10 +111,19 @@ func (p *publishes) ended(ctx context.Context, e jetstream.KeyValueEntry) {
|
||||
if err != nil || v.State != versions.Numbered {
|
||||
return
|
||||
}
|
||||
p.findings.raise(ctx, finding{kind: versions.PublishUnanswered, scope: health.ScopeModule, id: v.Name,
|
||||
summary: fmt.Sprintf("%s said neither published nor refused of %s@%s within %s of its build", v.Seat, v.Name, v.Version, p.bound),
|
||||
headline: "A registry seat did not answer a publish",
|
||||
explanation: "The builder built and numbered " + v.Name + " " + v.Version + " and asked " + v.Seat + " to publish it; its holder said nothing within the bound, so the build says done while the version may not be in the registry.",
|
||||
resolved: "That version is published, or a later version of " + v.Name + " is.",
|
||||
needs: "a look at the holder of " + v.Seat + " (its health and log); builder_publish_again asks the publish again."})
|
||||
p.unanswered(ctx, v, fmt.Sprintf("nothing within %s of its build", p.bound))
|
||||
}
|
||||
|
||||
// watchBounds watches the state bounds, handing each change to each: a watch that cannot start loses every bound's end,
|
||||
// so it is the finding family-not-heard naming the state until a later start watches it (review round 2).
|
||||
func watchBounds(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyValueEntry), w heard.Writer, errs chan<- error, logf func(string, ...any)) {
|
||||
const consequence = "a publish no holder answers, and a node holding the old form seven days, are not raised until the builder starts again"
|
||||
pw, err := settle(ctx, kv, each)
|
||||
if err != nil {
|
||||
process.NotWatched(logf, versions.BoundsBucket, err, consequence)
|
||||
w.NotHeard(versions.BoundsState, consequence, err)
|
||||
return
|
||||
}
|
||||
w.Heard(versions.BoundsState)
|
||||
go func() { errs <- follow(ctx, pw, each) }()
|
||||
}
|
||||
|
||||
@@ -366,6 +366,18 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) {
|
||||
if got, err := health.DecodeFinding(st.vals[unansweredKey]); err != nil || !strings.Contains(got.Summary, "neither published nor refused of m/program@0.1.4") {
|
||||
t.Fatalf("a bound ended unanswered: the finding = %+v, %v", got, err)
|
||||
}
|
||||
// Review round 2: a bound that cannot be written is publish-unanswered at once; a watch of bounds that cannot start
|
||||
// is family-not-heard naming the state.
|
||||
delete(st.vals, unansweredKey)
|
||||
pubs.arm(context.Background(), versions.Entry{Name: "m/program", Version: "0.1.4", Seat: "oci-artifact-registry"})
|
||||
if got, err := health.DecodeFinding(st.vals[unansweredKey]); err != nil || !strings.Contains(got.Summary, "its bound could not be written") {
|
||||
t.Fatalf("a bound not written: the finding = %+v, %v", got, err)
|
||||
}
|
||||
watchBounds(context.Background(), refusedWatch{err: errors.New("permissions violation")}, func(jetstream.KeyValueEntry) {}, f.notHeard(context.Background()), nil, t.Logf)
|
||||
if got, err := health.DecodeFinding(st.vals[heard.Key(delivery.BuilderModule, versions.BoundsState)]); err != nil || !strings.Contains(got.Summary, "permissions violation") {
|
||||
t.Fatalf("bounds not watched: the finding = %+v, %v", got, err)
|
||||
}
|
||||
delete(st.vals, heard.Key(delivery.BuilderModule, versions.BoundsState))
|
||||
malformed := &wakeMsg{subject: "mesh.seat.oci-artifact-registry.event.published", data: []byte(`{"version":"0.1.4"}`)}
|
||||
unreadKey := health.FindingKey(health.ScopeModule, "oci-artifact-registry", versions.SeatOutcomeUnread)
|
||||
if !h.heard(context.Background(), malformed) || !malformed.acked || st.vals[unreadKey] == nil {
|
||||
|
||||
Reference in New Issue
Block a user