Let the kept compositions' pass end at the stop (issue 1945, ADR 0545, review 3)
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/delivery-group group fix/1945-kept-compositions-follow-config held: a member is held: novox/mesh-controller@ff3cc47027e0 (merged without a passing check: …
mesh/check failed: integration: github.com/novox/mesh-controller/cmd/mesh-controller.TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine: with …
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/delivery-group group fix/1945-kept-compositions-follow-config held: a member is held: novox/mesh-controller@ff3cc47027e0 (merged without a passing check: …
mesh/check failed: integration: github.com/novox/mesh-controller/cmd/mesh-controller.TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine: with …
followKept went on composing every kept key, and the config consumer every touched node, after the controller was told to stop. Both now check the stop before each key and each node; the writes of an unfinished batch are refused at once, so the next controller takes them. Research 072: milestone 2, the composer reacting to the change of an input.
This commit is contained in:
@@ -137,6 +137,9 @@ func followKept(ctx context.Context, open *stores, node string, modules map[stri
|
|||||||
var errs []error
|
var errs []error
|
||||||
builds := map[string][]inventory.Build{}
|
builds := map[string][]inventory.Build{}
|
||||||
for _, k := range kept {
|
for _, k := range kept {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return wrote, errors.Join(append(errs, ctx.Err())...) // stopped: the rest is the next controller's
|
||||||
|
}
|
||||||
if _, read := builds[k.module]; !read {
|
if _, read := builds[k.module]; !read {
|
||||||
if builds[k.module], err = inv.Builds(ctx, k.module, 20); err != nil {
|
if builds[k.module], err = inv.Builds(ctx, k.module, 20); err != nil {
|
||||||
errs = append(errs, fmt.Errorf("%s's builds: %w", k.module, err))
|
errs = append(errs, fmt.Errorf("%s's builds: %w", k.module, err))
|
||||||
@@ -311,6 +314,10 @@ func followConfigOnce(ctx context.Context, open *stores) error {
|
|||||||
forgetResolved() // the next composition reads the values just written, not a read of the last 5 s
|
forgetResolved() // the next composition reads the values just written, not a read of the last 5 s
|
||||||
failed := false
|
failed := false
|
||||||
for node, modules := range now {
|
for node, modules := range now {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
failed = true // stopped: the writes are refused below and delivered to the next controller
|
||||||
|
break
|
||||||
|
}
|
||||||
if err := composerFor(open).compose(ctx, []string{node}, false); err != nil {
|
if err := composerFor(open).compose(ctx, []string{node}, false); err != nil {
|
||||||
fmt.Printf("config: %s is not composed after its config changed: %v\n", node, err)
|
fmt.Printf("config: %s is not composed after its config changed: %v\n", node, err)
|
||||||
failed = true
|
failed = true
|
||||||
@@ -321,7 +328,9 @@ func followConfigOnce(ctx context.Context, open *stores) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
for _, m := range msgs {
|
for _, m := range msgs {
|
||||||
if failed {
|
if failed && ctx.Err() != nil {
|
||||||
|
_ = m.Nak() // stopped: delivered at once to the next controller
|
||||||
|
} else if failed {
|
||||||
_ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again
|
_ = m.NakWithDelay(doneKeyRetry) // tried again: the bus delivers the writes again
|
||||||
} else {
|
} else {
|
||||||
acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||||
|
|||||||
Reference in New Issue
Block a user