Count the work the take-ins start in the background, so a test's mesh closes only once that work has ended (issue 1851, A5-A7; issue 1790, D3)
This commit is contained in:
@@ -94,7 +94,7 @@ func askFirstBuilds(ctx context.Context, open *stores, m link.SourceMoved, now t
|
|||||||
fmt.Printf(" the indexer's modules of %s cannot be read now (%v): they are registered when the builder's next "+
|
fmt.Printf(" the indexer's modules of %s cannot be read now (%v): they are registered when the builder's next "+
|
||||||
"write of one wakes the take-in\n", repository, err)
|
"write of one wakes the take-in\n", repository, err)
|
||||||
case len(modules) > 0:
|
case len(modules) > 0:
|
||||||
go takeNewModules(context.WithoutCancel(ctx), open, modules...)
|
inBackground(func() { takeNewModules(context.WithoutCancel(ctx), open, modules...) })
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -56,6 +56,8 @@ func aMeshOnTheStore(t *testing.T) *stores {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
t.Cleanup(open.Close)
|
t.Cleanup(open.Close)
|
||||||
|
// The work this mesh started in the background ends before its stores close (cleanups run last first).
|
||||||
|
t.Cleanup(background.Wait)
|
||||||
// A condition store of its own, held in memory (novox/hq to-be 45 §2): status leads with what is
|
// A condition store of its own, held in memory (novox/hq to-be 45 §2): status leads with what is
|
||||||
// open, and a mesh with no bus would otherwise read as one whose conditions cannot be read.
|
// open, and a mesh with no bus would otherwise read as one whose conditions cannot be read.
|
||||||
withConditionsInMemory(t)
|
withConditionsInMemory(t)
|
||||||
|
|||||||
@@ -200,10 +200,10 @@ func newModuleObservations(now time.Time) []conditions.Observation {
|
|||||||
// watchNewModules wakes the take-in of new modules for each module the builder or the indexer writes, until ctx ends, and
|
// watchNewModules wakes the take-in of new modules for each module the builder or the indexer writes, until ctx ends, and
|
||||||
// looks at every module once now. A bucket not there yet is said: the look at start and a first merge's word still work.
|
// looks at every module once now. A bucket not there yet is said: the look at start and a first merge's word still work.
|
||||||
func watchNewModules(ctx context.Context, open *stores) {
|
func watchNewModules(ctx context.Context, open *stores) {
|
||||||
go func() {
|
inBackground(func() {
|
||||||
answerHeldMerges(ctx, open.inventory)
|
answerHeldMerges(ctx, open.inventory)
|
||||||
takeNewModules(ctx, open)
|
takeNewModules(ctx, open)
|
||||||
}()
|
})
|
||||||
keys, closeKeys, err := openBuilderKeys()
|
keys, closeKeys, err := openBuilderKeys()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
|
fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
|
||||||
@@ -223,7 +223,7 @@ func watchNewModules(ctx context.Context, open *stores) {
|
|||||||
// wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
|
// wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
|
||||||
// draining.
|
// draining.
|
||||||
func wakeNewModule(ctx context.Context, open *stores, module string) {
|
func wakeNewModule(ctx context.Context, open *stores, module string) {
|
||||||
go func() {
|
inBackground(func() {
|
||||||
entries, err := open.inventory.Catalogued(ctx)
|
entries, err := open.inventory.Catalogued(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return
|
||||||
@@ -236,6 +236,19 @@ func wakeNewModule(ctx context.Context, open *stores, module string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
takeNewModules(ctx, open, module)
|
takeNewModules(ctx, open, module)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// background is the work the take-ins of keys and of announcements start beside the event that woke them, so a test
|
||||||
|
// ends only once the work its mesh started has.
|
||||||
|
var background sync.WaitGroup
|
||||||
|
|
||||||
|
// inBackground runs work beside the caller, counted in background.
|
||||||
|
func inBackground(work func()) {
|
||||||
|
background.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer background.Done()
|
||||||
|
work()
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -73,7 +73,7 @@ var keyTakes = struct {
|
|||||||
// the buckets and the records; a done key not recorded yet is taken in off the plans' hold, and its end asks the look.
|
// the buckets and the records; a done key not recorded yet is taken in off the plans' hold, and its end asks the look.
|
||||||
func settleFromKeys(ctx context.Context, open *stores, keys planKeyReads, p *inventory.Plan, tier []string) (bool, error) {
|
func settleFromKeys(ctx context.Context, open *stores, keys planKeyReads, p *inventory.Plan, tier []string) (bool, error) {
|
||||||
return settleKeys(ctx, open.inventory, keys, p, tier, func(module string, trunk sdkdelivery.Module, id string) {
|
return settleKeys(ctx, open.inventory, keys, p, tier, func(module string, trunk sdkdelivery.Module, id string) {
|
||||||
go takeKeyOffTheHold(context.WithoutCancel(ctx), open, module, trunk, id)
|
inBackground(func() { takeKeyOffTheHold(context.WithoutCancel(ctx), open, module, trunk, id) })
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -67,7 +67,7 @@ func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores)
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
if changed {
|
if changed {
|
||||||
go composeAnnouncedReads(context.WithoutCancel(ctx), open)
|
inBackground(func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user