diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index c864de3..51270c8 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -602,9 +602,13 @@ func (c *Conn) HandleCalled(key string, h CalledHandler) (func(), error) { } var mu sync.Mutex var stops []func() + stopped := false serve := func() { mu.Lock() defer mu.Unlock() + if stopped { + return + } for _, s := range stops { s() } @@ -618,14 +622,18 @@ func (c *Conn) HandleCalled(key string, h CalledHandler) (func(), error) { } } serve() - c.OnMembership(func(m Membership) { + // Its stop drops the membership handler too: a handler left behind would bind the stopped tool again + // on the module's next membership, and call a handler whose bundle was replaced (issue 1945). + unfollow := c.onMembership(func(m Membership) { if m.Module == module { serve() } }) return func() { + unfollow() mu.Lock() defer mu.Unlock() + stopped = true for _, s := range stops { s() } diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index b4c3af8..6492a39 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -174,6 +174,11 @@ func (c *child) write(m any) error { var stackLine = regexp.MustCompile(`^\s+at\s`) +// colour is a terminal's colour or style sequence, which node writes into a stack trace when its +// environment asks for colour (FORCE_COLOR): taken out before a line is judged or kept as a bundle's +// reason, so the reason reads the same in every environment. +var colour = regexp.MustCompile("\x1b\\[[0-9;]*m") + // Start launches one bundle and learns its tools. It fails when the child cannot be started or does // not complete the handshake. A child that exits later is started again on its next call. func Start(module, entry string, env []string, mesh Bus, logf func(string, ...any)) (*Launched, error) { @@ -245,9 +250,10 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an continue } logf("[%s] %s", module, line) - if !stackLine.MatchString(line) && !strings.HasPrefix(line, "Node.js v") { + plain := colour.ReplaceAllString(line, "") + if !stackLine.MatchString(plain) && !strings.HasPrefix(plain, "Node.js v") { saidMu.Lock() - lastSaid = strings.TrimSpace(line) + lastSaid = strings.TrimSpace(plain) saidMu.Unlock() } } @@ -456,14 +462,24 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an asking := func(method string, params any, timeout time.Duration) (json.RawMessage, error) { mu.Lock() - c := current + c, gone := current, stopped mu.Unlock() + if gone { + // A stopped bundle is never started again by a call: a process started now would be one + // nothing stops (issue 1945). + return nil, fmt.Errorf("%s's bundle is stopped; it answers no call", module) + } if c == nil { fresh, err := start() if err != nil { return nil, err } mu.Lock() + if stopped { + mu.Unlock() + _ = fresh.cmd.Process.Signal(syscall.SIGTERM) + return nil, fmt.Errorf("%s's bundle is stopped; it answers no call", module) + } if current == nil { current = fresh startedAt = time.Now() diff --git a/node-tools/internal/runtime/reload_test.go b/node-tools/internal/runtime/reload_test.go index 109a1e2..52599df 100644 --- a/node-tools/internal/runtime/reload_test.go +++ b/node-tools/internal/runtime/reload_test.go @@ -132,6 +132,7 @@ func fastSettle(t *testing.T) { } type reloadBed struct { + mesh *mt.Mesh nodeTools *bus.Conn asker *bus.Conn logs *syncLogs @@ -144,6 +145,7 @@ func newReloadBed(t *testing.T) *reloadBed { mesh := mt.New(t) mesh.Issue(t, mt.MembershipOf("ledger", "anchor", false, nil)) return &reloadBed{ + mesh: mesh, nodeTools: connect(t, "node-tools", "anchor"), asker: connect(t, "console", "workstation"), logs: &syncLogs{}, @@ -259,6 +261,27 @@ func TestAReplacedBundleIsReloaded(t *testing.T) { b.answers(t, "fourth") } +// A replaced bundle's old tool stays gone when the module's membership is issued again: a stopped tool +// leaves no membership handler behind to bind it once more, and a call on it starts no unmanaged second +// copy of the bundle (issue 1945, review round 2). +func TestAReplacedBundlesOldToolStaysGoneOnTheNextMembership(t *testing.T) { + b := newReloadBed(t) + place(t, b.entry, "first") + b.run(t) + b.answers(t, "first") + place(t, b.entry, "second") + b.answers(t, "second") + b.silent(t, "first") + + b.mesh.Issue(t, mt.MembershipOf("ledger", "anchor", false, nil)) + b.answers(t, "second") + // Long enough for the reissued membership to have been heard, before the old tool is asked. + time.Sleep(300 * time.Millisecond) + if got, err := call(t, b.asker, "ledger.first@anchor", map[string]any{}); err == nil { + t.Fatalf("ledger.first is served again after the membership was reissued, and answered %s\n%s", got, b.logs.All()) + } +} + // 1945/AC21: a bundle removed while served is no longer served, and its module says why. func TestARemovedBundleIsNoLongerServed(t *testing.T) { b := newReloadBed(t) diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index d7b013c..5cd47f9 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -480,8 +480,9 @@ func (r *runner) reload(module, what string) { old := r.of[module] r.of[module] = &loadedModule{served: old.served, seen: old.seen} r.mu.Unlock() - old.unload() + // Its events are refused first, so one heard while the old bundle stops is refused at once. r.events.dropModule(module) + old.unload() r.logf("[mesh-tools] %s's bundle %s; loading it again", module, what) fresh := r.load(old.served) diff --git a/node-tools/internal/runtime/watch.go b/node-tools/internal/runtime/watch.go index d51c347..7ae673b 100644 --- a/node-tools/internal/runtime/watch.go +++ b/node-tools/internal/runtime/watch.go @@ -61,26 +61,40 @@ func watchBundles(paths []string, changed func(), logf func(string, ...any)) (fu } // arm watches, for every entrypoint, its directory, or the nearest directory above it that exists. A -// watch already held on the same directory is kept (inotify answers its descriptor again). +// watch already held on the same directory is kept (inotify answers its descriptor again). A directory +// made below the watched one between its stat and its watch would be heard by nobody, so after each watch +// the path is looked at again, and a deeper directory that came to be is watched in turn. func (w *bundleWatch) arm() { for _, p := range w.paths { - dir := filepath.Dir(p) + target := filepath.Dir(p) + watched := "" for { - if info, err := os.Stat(dir); err == nil && info.IsDir() { + dir := nearestDir(target) + if dir == watched { break } - up := filepath.Dir(dir) - if up == dir { + wd, err := unix.InotifyAddWatch(w.fd, dir, watchMask) + if err != nil { + w.logf("[mesh-tools] cannot watch %s for %s's bundle: %v; a bundle placed or changed there is served only after a restart", dir, p, err) break } - dir = up + w.byWD[wd] = dir + watched = dir } - wd, err := unix.InotifyAddWatch(w.fd, dir, watchMask) - if err != nil { - w.logf("[mesh-tools] cannot watch %s for %s's bundle: %v; a bundle placed or changed there is served only after a restart", dir, p, err) - continue + } +} + +// nearestDir is dir when it is a directory, else the nearest directory above it. +func nearestDir(dir string) string { + for { + if info, err := os.Stat(dir); err == nil && info.IsDir() { + return dir } - w.byWD[wd] = dir + up := filepath.Dir(dir) + if up == dir { + return dir + } + dir = up } }