Forget a kept build when its key is written again, and retry data items on the composer's bound (review round 1)
mesh/delivery merged: source.main moved to novox/mesh-host@233bb23bbcb1
mesh/delivery-group group feat/data-items merged: every member is merged
mesh/check pass e031758e4a37 on 6c1f7cc7f50f | unit 4510 run, 4259 kept, 49 skipped | integration 4589 run, 4361 kept, 51 skipped | over bound: unit, …
mesh/delivery merged: source.main moved to novox/mesh-host@233bb23bbcb1
mesh/delivery-group group feat/data-items merged: every member is merged
mesh/check pass e031758e4a37 on 6c1f7cc7f50f | unit 4510 run, 4259 kept, 49 skipped | integration 4589 run, 4361 kept, 51 skipped | over bound: unit, …
The builder claims a done key again as a new revision, so a build kept as done is dropped on any write of its key. Data items that could not be worked out or written stay due and owe a pass after the growing bound, so a first write refused before the state or grant exists is tried again without waiting for an unrelated composition. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -6,6 +6,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
"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)
|
items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, scope)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " +
|
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
|
return
|
||||||
}
|
}
|
||||||
if len(unplaced) > 0 {
|
if len(unplaced) > 0 {
|
||||||
@@ -70,13 +71,13 @@ func (c *Composer) publishData(ctx context.Context) {
|
|||||||
c.mu.Unlock()
|
c.mu.Unlock()
|
||||||
wrote, removed, err := c.writeData(ctx, items, digests)
|
wrote, removed, err := c.writeData(ctx, items, digests)
|
||||||
if err != nil {
|
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 +
|
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
|
return
|
||||||
}
|
}
|
||||||
|
c.mu.Lock()
|
||||||
|
c.dataTries = 0
|
||||||
|
c.mu.Unlock()
|
||||||
c.sayData("")
|
c.sayData("")
|
||||||
if wrote+removed > 0 {
|
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",
|
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.
|
// sayData says a line about the data items once until it changes; "" clears it.
|
||||||
func (c *Composer) sayData(line string) {
|
func (c *Composer) sayData(line string) {
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
|
|||||||
@@ -7,10 +7,13 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||||
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"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.
|
// 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
|
// 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 `<node>.<module>`, with its composition's digest: written when the module's
|
// module's data items in its state data, key `<node>.<module>`, 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
|
// 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) {
|
func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
|
||||||
item := func(module, id, path string) catalogue.PlacedDataItem {
|
item := func(module, id, path string) catalogue.PlacedDataItem {
|
||||||
return catalogue.PlacedDataItem{Module: module, ID: id, Class: "valuable", Path: path, Measure: catalogue.MeasureWalk}
|
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}`)}}
|
"n1.gone": []byte(`{"v":1}`), "n2.web": []byte(`{"v":1}`)}}
|
||||||
var items map[string][]catalogue.PlacedDataItem
|
var items map[string][]catalogue.PlacedDataItem
|
||||||
var asked int
|
var asked int
|
||||||
|
var itemsErr error
|
||||||
setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, error) {
|
setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, error) {
|
||||||
asked++
|
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.
|
// 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.*") {
|
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
|
items map[string][]catalogue.PlacedDataItem
|
||||||
digests map[string]string // composed key to digest handed
|
digests map[string]string // composed key to digest handed
|
||||||
fail error
|
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
|
puts []string
|
||||||
deletes []string
|
deletes []string
|
||||||
asked int
|
asked int
|
||||||
@@ -106,23 +113,26 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
|
|||||||
{name: "a pass that changes no module's items writes nothing", due: true,
|
{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")}},
|
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},
|
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")}},
|
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,
|
{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")}},
|
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{
|
want: map[string]delivery.EngineData{
|
||||||
"n1.web": {Module: "web", Digest: "sha256:w2", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db2")}}}},
|
"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,
|
{name: "a module that left the node is deleted", due: true,
|
||||||
items: map[string][]catalogue.PlacedDataItem{"oldie": {item("oldie", "x", "/srv/oldie")}},
|
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) {
|
t.Run(step.name, func(t *testing.T) {
|
||||||
store.mu.Lock()
|
store.mu.Lock()
|
||||||
store.puts, store.deletes, store.fail = nil, nil, step.fail
|
store.puts, store.deletes, store.fail = nil, nil, step.fail
|
||||||
store.mu.Unlock()
|
store.mu.Unlock()
|
||||||
items = step.items
|
items, itemsErr = step.items, step.refused
|
||||||
|
c.paid() // the pass before composed: nothing owed
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
c.handed = step.digests
|
c.handed = step.digests
|
||||||
c.dataDue = c.dataDue || step.due
|
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,
|
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)
|
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 {
|
for key, want := range step.want {
|
||||||
got, err := delivery.DecodeEngineData(store.kv[key])
|
got, err := delivery.DecodeEngineData(store.kv[key])
|
||||||
if err != nil || got.Node != "n1" || got.Module != want.Module || got.Digest != want.Digest ||
|
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 {
|
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)
|
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)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
// (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.
|
// data hold, read once per run, nil before; the loop's alone.
|
||||||
dataDue bool
|
dataDue bool
|
||||||
|
dataTries int // data runs failed in a row, for oweData's bound
|
||||||
dataSaid string
|
dataSaid string
|
||||||
dataScope frombus.Scope
|
dataScope frombus.Scope
|
||||||
dataKept map[string][]byte
|
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
|
// 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.
|
// 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 {
|
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 {
|
if u.Deleted {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user