DLV-B6: a branch's end removes its tagged versions, acknowledged once every seat said removed; the checker's tag-wanted is kept in tag-asks
The builder hears the forge bridge's stream on a durable consumer: merged, a pull's closed and branch-deleted ask remove of each tagged version still held, and the bus gives the message again each minute until all are removed; past 15 minutes its finding tag-removal-not-done names the seats. The builder gains the state findings (ADR 0564 §2). Every forge message also wakes a floor whose bridge_tags was silent. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -0,0 +1,142 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
|
||||
"mesh-delivery-builder/internal/versions"
|
||||
)
|
||||
|
||||
// forgeStream is the forge bridge's stream `forge` as the bus names it, where the builder's durable consumer hears a
|
||||
// branch's end (design 51 *Tagged versions*).
|
||||
const forgeStream = "FORGE"
|
||||
|
||||
// forgeBranchDeleted is the forge bridge's kind for a branch's end (delivery.ForgeBranchDeleted in the group's mesh-sdk).
|
||||
const forgeBranchDeleted = "branch-deleted"
|
||||
|
||||
// checkerAsksStream is the checker's family `asks` as the bus names it, and tagWantedSubject its event tag-wanted
|
||||
// (design 51 *Tagged versions*, DLV-C3).
|
||||
const (
|
||||
checkerAsksStream = "ASKS"
|
||||
tagWantedSubject = "mesh.mod." + delivery.CheckerModule + ".event.tag-wanted"
|
||||
)
|
||||
|
||||
// forgeEnd is the part of a forge message the builder reads.
|
||||
type forgeEnd struct {
|
||||
Kind string `json:"kind"`
|
||||
Repository string `json:"repository"`
|
||||
Branch string `json:"branch"`
|
||||
Action string `json:"action"`
|
||||
}
|
||||
|
||||
// ended says whether the message ends a branch: merged, a pull closed, or branch-deleted.
|
||||
func (f forgeEnd) ended() bool {
|
||||
switch f.Kind {
|
||||
case delivery.ForgeMerged, forgeBranchDeleted:
|
||||
return f.Branch != ""
|
||||
case delivery.ForgePull:
|
||||
return f.Action == delivery.PullClosed && f.Branch != ""
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// branchEnds hears the forge bridge's stream: every message wakes a floor whose bridge_tags was silent, and a branch's
|
||||
// end removes its tagged versions, acknowledged only once all are removed.
|
||||
type branchEnds struct {
|
||||
acts versions.Acts
|
||||
findings *findings
|
||||
tagAsks func() jetstream.KeyValue
|
||||
wake func(ctx context.Context)
|
||||
logf func(string, ...any)
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func (h *branchEnds) heard(ctx context.Context, m jetstream.Msg) {
|
||||
if h.wake != nil {
|
||||
h.wake(ctx)
|
||||
}
|
||||
var f forgeEnd
|
||||
if err := json.Unmarshal(m.Data(), &f); err != nil || !f.ended() {
|
||||
h.ack(m)
|
||||
return
|
||||
}
|
||||
end, err := h.acts.BranchEnded(ctx, f.Repository, f.Branch)
|
||||
if err != nil {
|
||||
h.logf("the end of %s %s: the versions cannot be read, heard again: %v", f.Repository, f.Branch, err)
|
||||
h.again(m)
|
||||
return
|
||||
}
|
||||
id := f.Repository + "/" + f.Branch
|
||||
if end.Done {
|
||||
h.findings.clear(ctx, versions.RemovalNotDone, health.ScopeModule, id)
|
||||
if kv := h.tagAsks(); kv != nil {
|
||||
if err := kv.Delete(ctx, versions.TagAskKey(f.Repository, f.Branch)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
h.logf("the tag ask of %s %s stays in %s: %v", f.Repository, f.Branch, versions.TagAsksState, err)
|
||||
}
|
||||
}
|
||||
h.ack(m)
|
||||
return
|
||||
}
|
||||
since := h.now()
|
||||
if md, err := m.Metadata(); err == nil {
|
||||
since = md.Timestamp
|
||||
}
|
||||
if h.now().Sub(since) >= versions.RemovalFindingAfter {
|
||||
h.findings.raise(ctx, finding{kind: versions.RemovalNotDone, scope: health.ScopeModule, id: id,
|
||||
summary: end.Finding(f.Repository, f.Branch, since),
|
||||
headline: "A branch's tagged versions are not removed",
|
||||
explanation: "The branch ended, and the registry seats did not say removed for every version tagged for it within 15 minutes.",
|
||||
resolved: "Every seat said removed.",
|
||||
needs: "Look at the seats named: " + strings.Join(end.Seats(), ", ")})
|
||||
}
|
||||
h.logf("the end of %s %s: remove asked of %d tagged versions, heard again in a minute", f.Repository, f.Branch, len(end.Left))
|
||||
h.again(m)
|
||||
}
|
||||
|
||||
func (h *branchEnds) ack(m jetstream.Msg) {
|
||||
if err := m.Ack(); err != nil {
|
||||
h.logf("acknowledging %s: %v", m.Subject(), err)
|
||||
}
|
||||
}
|
||||
|
||||
// again leaves the message for the bus to give again in a minute: the retry is the bus's redelivery, not a timer here.
|
||||
func (h *branchEnds) again(m jetstream.Msg) {
|
||||
if err := m.NakWithDelay(versions.RemoveAgainAfter); err != nil {
|
||||
h.logf("leaving %s for the bus to give again: %v", m.Subject(), err)
|
||||
}
|
||||
}
|
||||
|
||||
// tagWanted takes the checker's tag-wanted in: written to tag-asks as builder_tag writes it, answered by its watch.
|
||||
func tagWanted(ctx context.Context, kv jetstream.KeyValue, m jetstream.Msg, by string, now time.Time, logf func(string, ...any)) {
|
||||
var t versions.TagWanted
|
||||
err := json.Unmarshal(m.Data(), &t)
|
||||
var key string
|
||||
var value []byte
|
||||
if err == nil {
|
||||
key, value, err = versions.TagWantedAsk(t, by, now)
|
||||
}
|
||||
if err != nil {
|
||||
logf("%s holds what is no tag-wanted, left: %v", m.Subject(), err)
|
||||
_ = m.Ack()
|
||||
return
|
||||
}
|
||||
if kv == nil {
|
||||
logf("tag-wanted of %s %s is heard again: %s is not on the bus yet", t.Repository, t.Branch, versions.TagAsksState)
|
||||
_ = m.NakWithDelay(versions.RemoveAgainAfter)
|
||||
return
|
||||
}
|
||||
if _, err := kv.Put(ctx, key, value); err != nil {
|
||||
logf("tag-wanted of %s %s is heard again: %v", t.Repository, t.Branch, err)
|
||||
_ = m.Nak()
|
||||
return
|
||||
}
|
||||
_ = m.Ack()
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
|
||||
// findingsBucket is the builder's findings (novox/hq ADR 0564 §2): what it sees wrong now, written at its raise and
|
||||
// deleted at its clear, which the condition keeper raises and clears.
|
||||
var findingsBucket = health.FindingsBucket(delivery.BuilderModule)
|
||||
|
||||
// findings writes the builder's findings into its state, once bound; until then each is said in the log only.
|
||||
type findings struct {
|
||||
kv *process.AwaitedBucket
|
||||
logf func(string, ...any)
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
// finding is one of the builder's findings: its kind, what it is about, and its words.
|
||||
type finding struct {
|
||||
kind, scope, id string
|
||||
summary, headline, explanation, resolved, needs string
|
||||
severity health.Severity
|
||||
}
|
||||
|
||||
func (f *findings) bucket() jetstream.KeyValue {
|
||||
if f == nil || f.kv == nil {
|
||||
return nil
|
||||
}
|
||||
return f.kv.KV()
|
||||
}
|
||||
|
||||
// raise writes fd; it is written again only when its words changed, so the keeper's condition moves only then.
|
||||
func (f *findings) raise(ctx context.Context, fd finding) {
|
||||
key := health.FindingKey(fd.scope, fd.id, fd.kind)
|
||||
kv := f.bucket()
|
||||
if kv == nil {
|
||||
f.logf("finding %s (not written, %s is not on the bus yet): %s", key, findingsBucket, fd.summary)
|
||||
return
|
||||
}
|
||||
if e, err := kv.Get(ctx, key); err == nil {
|
||||
if old, err := health.DecodeFinding(e.Value()); err == nil && old.Summary == fd.summary {
|
||||
return
|
||||
}
|
||||
}
|
||||
at := f.now().UTC()
|
||||
sev := fd.severity
|
||||
if sev == "" {
|
||||
sev = health.Warning
|
||||
}
|
||||
value, err := health.EncodeFinding(health.Finding{Key: key, Kind: fd.kind, Subject: health.Subject{Scope: fd.scope, ID: fd.id},
|
||||
Severity: sev, Summary: fd.summary, Headline: fd.headline, Explanation: fd.explanation, Resolved: fd.resolved, Needs: fd.needs,
|
||||
Evidence: []health.Evidence{{At: at, Said: fd.summary}}, Source: "builder." + fd.kind, Observer: delivery.BuilderModule, At: at})
|
||||
if err == nil {
|
||||
_, err = kv.Put(ctx, key, value)
|
||||
}
|
||||
if err != nil {
|
||||
f.logf("the finding %s cannot be written: %v", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
// clear deletes the finding of kind about scope and id; one not there is nothing to clear.
|
||||
func (f *findings) clear(ctx context.Context, kind, scope, id string) {
|
||||
kv := f.bucket()
|
||||
if kv == nil {
|
||||
return
|
||||
}
|
||||
key := health.FindingKey(scope, id, kind)
|
||||
if _, err := kv.Get(ctx, key); errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return
|
||||
}
|
||||
if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
f.logf("the finding %s cannot be cleared: %v", key, err)
|
||||
}
|
||||
}
|
||||
@@ -292,7 +292,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
})
|
||||
// A tag ask or a branch's removal by hand (builder_tag, builder_untag) is written to tag-asks and answered there
|
||||
// (design 51 *Tagged versions*); the checker's tag-wanted and the branch's end join it on their consumers (DLV-B6).
|
||||
process.AwaitBucket(ctx, js, conn, versions.TagAsksBucket, "builder_tag and builder_untag are not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) {
|
||||
tagAsksKV := process.AwaitBucket(ctx, js, conn, versions.TagAsksBucket, "builder_tag and builder_untag are not answered", logger.Printf)
|
||||
tagAsksKV.Ready(func(kv jetstream.KeyValue) {
|
||||
each := func(e jetstream.KeyValueEntry) { tagAsked(ctx, acts, kv, e, logger.Printf) }
|
||||
if tw, err := settle(ctx, kv, each); err != nil {
|
||||
process.NotWatched(logger.Printf, versions.TagAsksBucket, err, "builder_tag and builder_untag are not answered until the builder starts again")
|
||||
@@ -310,6 +311,34 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
go func() { errs <- follow(ctx, qw, eachRead) }()
|
||||
}
|
||||
})
|
||||
// A branch's end removes its tagged versions (design 51 *Tagged versions*, DLV-B6): the forge bridge's merged, a pull's
|
||||
// closed or branch-deleted on the builder's durable consumer of its stream; each forge message also wakes a floor whose
|
||||
// bridge_tags was silent. The checker's tag-wanted is kept in tag-asks. A consumer not there yet is said.
|
||||
found := &findings{kv: process.AwaitBucket(ctx, js, conn, findingsBucket, "the builder's findings are said in its log only", logger.Printf), logf: logger.Printf, now: time.Now}
|
||||
ends := &branchEnds{acts: acts, findings: found, tagAsks: tagAsksKV.KV, logf: logger.Printf, now: time.Now,
|
||||
wake: func(ctx context.Context) {
|
||||
(&floorWake{b: b, waits: floorWaits, logf: logger.Printf}).wake(ctx, versions.WakeForge, "", "")
|
||||
}}
|
||||
for _, c := range []struct {
|
||||
stream, what string
|
||||
hear func(jetstream.Msg)
|
||||
}{
|
||||
{forgeStream, "a branch's end", func(m jetstream.Msg) { ends.heard(ctx, m) }},
|
||||
{checkerAsksStream, "the checker's tag-wanted", func(m jetstream.Msg) {
|
||||
tagWanted(ctx, tagAsksKV.KV(), m, delivery.CheckerModule, time.Now(), logger.Printf)
|
||||
}},
|
||||
} {
|
||||
cons, err := cred.EventConsumer(ctx, js, c.stream)
|
||||
if err != nil {
|
||||
logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v", cred.Node, cred.Module, c.stream, c.what, err)
|
||||
continue
|
||||
}
|
||||
cc, err := process.HearEvents(cons, c.hear, logger.Printf)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer cc.Stop()
|
||||
}
|
||||
logger.Printf("started on %s", cred.Node)
|
||||
go func() { errs <- follow(ctx, w, each) }()
|
||||
select {
|
||||
|
||||
@@ -27,6 +27,7 @@ import (
|
||||
// issue 774).
|
||||
func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) {
|
||||
t.Run("a floor's wake is heard on the events consumer (design 51 The floor)", aFloorsWakeIsHeard)
|
||||
t.Run("a branch's end is acknowledged once its tagged versions are removed (design 51 Tagged versions)", aBranchsEndIsHeard)
|
||||
set := &buildrun.ToolchainSet{}
|
||||
var logs bytes.Buffer
|
||||
logger := log.New(&logs, "", 0)
|
||||
@@ -85,12 +86,20 @@ type wakeMsg struct {
|
||||
data []byte
|
||||
acked bool
|
||||
naked bool
|
||||
delayed time.Duration
|
||||
}
|
||||
|
||||
func (m *wakeMsg) Subject() string { return m.subject }
|
||||
func (m *wakeMsg) Data() []byte { return m.data }
|
||||
func (m *wakeMsg) Ack() error { m.acked = true; return nil }
|
||||
func (m *wakeMsg) Nak() error { m.naked = true; return nil }
|
||||
func (m *wakeMsg) NakWithDelay(d time.Duration) error {
|
||||
m.delayed = d
|
||||
return nil
|
||||
}
|
||||
func (m *wakeMsg) Metadata() (*jetstream.MsgMetadata, error) {
|
||||
return nil, errors.New("no metadata")
|
||||
}
|
||||
func (m *wakeMsg) Headers() nats.Header {
|
||||
return nil
|
||||
}
|
||||
@@ -144,3 +153,44 @@ func (noVersions) Update(context.Context, string, []byte, uint64) (uint64, error
|
||||
return 0, errors.New("read only")
|
||||
}
|
||||
func (noVersions) KeysOf(context.Context, string) ([]string, error) { return nil, nil }
|
||||
|
||||
func aBranchsEndIsHeard(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
f forgeEnd
|
||||
want bool
|
||||
}{
|
||||
{forgeEnd{Kind: delivery.ForgeMerged, Repository: "o/r", Branch: "feat/x"}, true},
|
||||
{forgeEnd{Kind: forgeBranchDeleted, Repository: "o/r", Branch: "feat/x"}, true},
|
||||
{forgeEnd{Kind: delivery.ForgePull, Repository: "o/r", Branch: "feat/x", Action: delivery.PullClosed}, true},
|
||||
{forgeEnd{Kind: delivery.ForgePull, Repository: "o/r", Branch: "feat/x", Action: delivery.PullOpened}, false},
|
||||
{forgeEnd{Kind: delivery.ForgeMerged, Repository: "o/r"}, false},
|
||||
} {
|
||||
if got := c.f.ended(); got != c.want {
|
||||
t.Errorf("%+v ended = %v, want %v", c.f, got, c.want)
|
||||
}
|
||||
}
|
||||
var logs bytes.Buffer
|
||||
woken := 0
|
||||
h := &branchEnds{acts: versions.Acts{Numberer: &versions.Numberer{Versions: versions.Versions{Store: noVersions{}}}}, findings: &findings{logf: t.Logf},
|
||||
tagAsks: func() jetstream.KeyValue { return nil }, wake: func(context.Context) { woken++ }, logf: log.New(&logs, "", 0).Printf, now: time.Now}
|
||||
open := &wakeMsg{subject: "mesh.mod.mesh-delivery-forge-bridge.event.forge.o.r.pull", data: []byte(`{"kind":"pull","repository":"o/r","branch":"feat/x","action":"opened"}`)}
|
||||
h.heard(context.Background(), open)
|
||||
merged := &wakeMsg{subject: "mesh.mod.mesh-delivery-forge-bridge.event.forge.o.r.merged", data: []byte(`{"kind":"merged","repository":"o/r","branch":"feat/x"}`)}
|
||||
h.heard(context.Background(), merged)
|
||||
if !open.acked || !merged.acked || woken != 2 {
|
||||
t.Errorf("an open pull acked %v, a merge with nothing tagged acked %v, floors woken %d times", open.acked, merged.acked, woken)
|
||||
}
|
||||
unreadable := &wakeMsg{subject: merged.subject, data: merged.data}
|
||||
h.acts.Numberer.Versions.Store = failingVersions{}
|
||||
h.heard(context.Background(), unreadable)
|
||||
if unreadable.acked || unreadable.delayed != versions.RemoveAgainAfter {
|
||||
t.Errorf("a branch's end the versions cannot answer: acked %v, given again after %v", unreadable.acked, unreadable.delayed)
|
||||
}
|
||||
}
|
||||
|
||||
// failingVersions is a state versions that cannot be read.
|
||||
type failingVersions struct{ noVersions }
|
||||
|
||||
func (failingVersions) KeysOf(context.Context, string) ([]string, error) {
|
||||
return nil, errors.New("down")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user