diff --git a/internal/selfcompose/data.go b/internal/selfcompose/data.go index ab78558..b21bcdb 100644 --- a/internal/selfcompose/data.go +++ b/internal/selfcompose/data.go @@ -6,6 +6,7 @@ import ( "fmt" "sort" "strings" + "time" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/compose" @@ -59,7 +60,7 @@ func (c *Composer) publishData(ctx context.Context) { items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, scope) if err != nil { c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " + - delivery.EngineDataBucket + " holds stands: " + err.Error()) + delivery.EngineDataBucket + " holds stands; " + c.oweData() + ": " + err.Error()) return } if len(unplaced) > 0 { @@ -70,13 +71,13 @@ func (c *Composer) publishData(ctx context.Context) { c.mu.Unlock() wrote, removed, err := c.writeData(ctx, items, digests) if err != nil { - c.mu.Lock() - c.dataDue = true // the next pass tries again, whatever it hands - c.mu.Unlock() c.sayData("the data items of this node's modules are not all kept in " + delivery.EngineDataBucket + - ", and are written again at the next composition: " + err.Error()) + "; " + c.oweData() + ": " + err.Error()) return } + c.mu.Lock() + c.dataTries = 0 + c.mu.Unlock() c.sayData("") if wrote+removed > 0 { c.env.Say(fmt.Sprintf("the data items of this node's modules: %d key(s) of %s written, %d deleted", @@ -84,6 +85,20 @@ func (c *Composer) publishData(ctx context.Context) { } } +// oweData keeps the data items due and has the composer's loop make a pass again after the growing bound (LaterWait), as +// a pass refused for a bus that did not answer is: the first write after a delivery may be refused while the state or +// its grant is not on the bus yet, and nothing else would try again until an unrelated composition changed. It answers +// the words that say when. +func (c *Composer) oweData() string { + c.mu.Lock() + defer c.mu.Unlock() + c.dataDue = true + c.wait = LaterWait(c.dataTries) + c.dataTries++ + c.owed.Store(true) + return fmt.Sprintf("tried again at %s (try %d)", time.Now().Add(c.wait).Format("15:04:05"), c.dataTries) +} + // sayData says a line about the data items once until it changes; "" clears it. func (c *Composer) sayData(line string) { c.mu.Lock() diff --git a/internal/selfcompose/data_test.go b/internal/selfcompose/data_test.go index a676da9..796fc74 100644 --- a/internal/selfcompose/data_test.go +++ b/internal/selfcompose/data_test.go @@ -7,10 +7,13 @@ import ( "strings" "sync" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/agent" ) // fakeData is the engine's state data as a map, counting the writes and deletes made. @@ -62,7 +65,8 @@ func (f *fakeData) Keys(_ context.Context, pattern string) ([]string, error) { // ADR 0577 clause 1 (and its check in clause 5): after a pass that composed something new, the engine keeps each placed // module's data items in its state data, key `.`, with its composition's digest: written when the module's // items change, not written when they do not, and deleted when the module leaves the node; a key of another node is never -// touched, a pass that handed nothing new reads nothing, and a write that failed is tried again at the next pass. +// touched, a pass that handed nothing new reads nothing, and items that could not be worked out or written are tried +// again with no new composition, after the composer's growing bound (the loop's owed pass). func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { item := func(module, id, path string) catalogue.PlacedDataItem { return catalogue.PlacedDataItem{Module: module, ID: id, Class: "valuable", Path: path, Measure: catalogue.MeasureWalk} @@ -72,9 +76,10 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { "n1.gone": []byte(`{"v":1}`), "n2.web": []byte(`{"v":1}`)}} var items map[string][]catalogue.PlacedDataItem var asked int + var itemsErr error setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, error) { asked++ - return items, nil, nil + return items, nil, itemsErr }) // The engine's grant lets it read its own node's keys, as the listing at a run's first pass does, and no other's. if !grantAllows(t, "n1", delivery.EngineDataBucket, "n1.*") || grantAllows(t, "n1", delivery.EngineDataBucket, "n2.*") { @@ -89,6 +94,8 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { items map[string][]catalogue.PlacedDataItem digests map[string]string // composed key to digest handed fail error + refused error // what working out the items says + owed time.Duration // the bound the loop waits before trying again; 0 when nothing is owed puts []string deletes []string asked int @@ -106,23 +113,26 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { {name: "a pass that changes no module's items writes nothing", due: true, items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}}, digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"}, asked: 2}, - {name: "a write that fails is tried again at the next pass", due: true, fail: errors.New("the bus said no"), + {name: "items that cannot be worked out are tried again after the bound", due: true, refused: errors.New("no answer"), + digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 3, owed: LaterWait(0)}, + {name: "a write that fails is tried again with no new composition, after a longer bound", due: false, fail: errors.New("the bus said no"), items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}}, - digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 3}, + digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 4, owed: LaterWait(1)}, {name: "a module whose composition changed its items is written again", due: false, items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}}, - digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 4, + digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 5, want: map[string]delivery.EngineData{ "n1.web": {Module: "web", Digest: "sha256:w2", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db2")}}}}, {name: "a module that left the node is deleted", due: true, items: map[string][]catalogue.PlacedDataItem{"oldie": {item("oldie", "x", "/srv/oldie")}}, - digests: map[string]string{nodeKey: "sha256:n1"}, deletes: []string{"n1.web"}, asked: 5}, + digests: map[string]string{nodeKey: "sha256:n1"}, deletes: []string{"n1.web"}, asked: 6}, } { t.Run(step.name, func(t *testing.T) { store.mu.Lock() store.puts, store.deletes, store.fail = nil, nil, step.fail store.mu.Unlock() - items = step.items + items, itemsErr = step.items, step.refused + c.paid() // the pass before composed: nothing owed c.mu.Lock() c.handed = step.digests c.dataDue = c.dataDue || step.due @@ -133,6 +143,9 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { t.Fatalf("wrote %v and deleted %v, items asked %d times; want %v, %v, %d", store.puts, store.deletes, asked, step.puts, step.deletes, step.asked) } + if wait, _ := c.waitNow(); c.owed.Load() != (step.owed > 0) || wait != step.owed { + t.Errorf("owed a pass: %v after %s; want after %s", c.owed.Load(), wait, step.owed) + } for key, want := range step.want { got, err := delivery.DecodeEngineData(store.kv[key]) if err != nil || got.Node != "n1" || got.Module != want.Module || got.Digest != want.Digest || @@ -170,4 +183,13 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { if reads[doneKey] != 1 || reads[runningKey] != 3 { t.Errorf("the builds were read %v times over the bus; want the done one once, the running one each time", reads) } + // A done key the builder claims again (a gone step, issue 1937) and lands done again is read again, not served kept. + again, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: delivery.ManifestDigest([]byte("{}"))}) + b.mu.Lock() + b.kv[delivery.BuilderBuildsBucket+"/"+doneKey] = again + b.mu.Unlock() + r.c.takeBuildDone(agent.Update{Key: doneKey, Value: running}) + if got, err := r.Build(context.Background(), "web", fp); err != nil || got.Manifest != delivery.ManifestDigest([]byte("{}")) { + t.Errorf("a done build written again is served as %+v (%v), not read again", got, err) + } } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index a8d7b4d..61fb4f7 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -156,6 +156,7 @@ type Composer struct { // (publishData); dataSaid is the last line said of them. Under mu. dataKept is what this node's keys of the state // data hold, read once per run, nil before; the loop's alone. dataDue bool + dataTries int // data runs failed in a row, for oweData's bound dataSaid string dataScope frombus.Scope dataKept map[string][]byte @@ -309,6 +310,11 @@ func (c *Composer) inputs() []input { // takeBuildDone says whether a build that landed is one this node composes from: done, of a module it wants at that // build or whose trunk's build it is. Every other build, and a build not done, moves nothing. func (c *Composer) takeBuildDone(u agent.Update) bool { + // Any write or delete of a build's key forgets the build kept as done: the builder claims a done key again as a new + // revision (a gone step, issue 1937), which lands done with other artifacts. + c.mu.Lock() + delete(c.builds, u.Key) + c.mu.Unlock() if u.Deleted { return false }