From 3e3f0eb3eaeda7802fa8fc0cca84a64c7b26abc8 Mon Sep 17 00:00:00 2001 From: Guillaume Lours Date: Wed, 9 Sep 2026 09:34:15 +0200 Subject: [PATCH] fix(watch): coalesce rebuilds triggered during a burst of file changes compose watch triggered one rebuild per debounced batch, so a burst of edits arriving faster than the debounce window (e.g. from a coding agent editing several files) queued up several rebuilds in a row instead of a single consolidated one. Rebuilds now run asynchronously through a small scheduler that coalesces requests received while a rebuild is in flight into one trailing rebuild, without adding latency to an isolated edit. This also closes a race where a plain restart could run concurrently with an asynchronous rebuild of the same service. Signed-off-by: Guillaume Lours --- pkg/compose/rebuild_scheduler.go | 103 +++++++++ pkg/compose/rebuild_scheduler_test.go | 311 ++++++++++++++++++++++++++ pkg/compose/watch.go | 39 +++- pkg/compose/watch_test.go | 57 ++++- 4 files changed, 495 insertions(+), 15 deletions(-) create mode 100644 pkg/compose/rebuild_scheduler.go create mode 100644 pkg/compose/rebuild_scheduler_test.go diff --git a/pkg/compose/rebuild_scheduler.go b/pkg/compose/rebuild_scheduler.go new file mode 100644 index 0000000000..4ecec4901c --- /dev/null +++ b/pkg/compose/rebuild_scheduler.go @@ -0,0 +1,103 @@ +/* + + Copyright 2020 Docker Compose CLI authors + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "sync" + + "github.com/docker/compose/v5/pkg/utils" +) + +// rebuildFunc performs a rebuild for the given services. +type rebuildFunc func(services []string) error + +// rebuildScheduler coalesces rebuild requests received while a rebuild is +// already running into a single trailing rebuild, so that a burst of file +// change events never queues up more than one extra rebuild. +type rebuildScheduler struct { + rebuild rebuildFunc + + mu sync.Mutex + building bool + active utils.Set[string] // services being rebuilt by the current run, if any + pending utils.Set[string] // services queued for the next trailing rebuild + wg sync.WaitGroup +} + +func newRebuildScheduler(rebuild rebuildFunc) *rebuildScheduler { + return &rebuildScheduler{ + rebuild: rebuild, + pending: utils.NewSet[string](), + } +} + +// Request asks for a rebuild of services. It never blocks on the rebuild +// itself: the services are always merged into the pending set, and the run +// loop is (re)started only if it isn't already draining it. +// +// The decision must be made synchronously, under the lock, so that requests +// racing with the run loop's completion are never lost nor cause two +// rebuilds to run concurrently. +func (s *rebuildScheduler) Request(services []string) { + s.mu.Lock() + defer s.mu.Unlock() + s.pending.AddAll(services...) + if s.building { + return + } + s.building = true + s.wg.Add(1) + go s.run() +} + +// InFlightOrPending reports whether a rebuild for the given service is +// currently running or queued for the next trailing rebuild. Callers use +// this to avoid racing a plain restart against a rebuild started by an +// earlier, still-running batch. +func (s *rebuildScheduler) InFlightOrPending(service string) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.active.Has(service) || s.pending.Has(service) +} + +// run drains the pending set into a rebuild, looping to pick up whatever +// accumulated while that rebuild was in flight, until nothing is left. +func (s *rebuildScheduler) run() { + defer s.wg.Done() + for { + s.mu.Lock() + s.active = s.pending + s.pending = utils.NewSet[string]() + services := s.active.Elements() + s.mu.Unlock() + + _ = s.rebuild(services) + + s.mu.Lock() + s.active = nil + if len(s.pending) == 0 { + s.building = false + s.mu.Unlock() + return + } + s.mu.Unlock() + } +} + +// Wait blocks until the scheduler is idle: no rebuild is running and nothing +// is pending. +func (s *rebuildScheduler) Wait() { + s.wg.Wait() +} diff --git a/pkg/compose/rebuild_scheduler_test.go b/pkg/compose/rebuild_scheduler_test.go new file mode 100644 index 0000000000..4cee27e607 --- /dev/null +++ b/pkg/compose/rebuild_scheduler_test.go @@ -0,0 +1,311 @@ +/* + + Copyright 2020 Docker Compose CLI authors + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "errors" + "sort" + "sync" + "sync/atomic" + "testing" + "time" + + "gotest.tools/v3/assert" +) + +// fakeRebuilder is a deterministic, channel-driven stand-in for the real +// `s.rebuild` call. Every invocation: +// - records the requested services (sorted, for easy comparison), +// - blocks until the test sends a value (nil or an error) on `release`, +// - fails the test via t.Errorf (safe to call from any goroutine) if it +// is ever entered while another invocation hasn't returned yet, which +// is exactly the "no concurrent rebuild" property under test. +type fakeRebuilder struct { + t *testing.T + + running int32 // atomic: 1 while an invocation is in flight + + started chan []string // signaled with sorted services each time rebuild() is entered + release chan error // test sends here to let the current invocation return + + mu sync.Mutex + calls [][]string // sorted services for every completed invocation, in order +} + +func newFakeRebuilder(t *testing.T) *fakeRebuilder { + t.Helper() + return &fakeRebuilder{ + t: t, + started: make(chan []string), + release: make(chan error), + } +} + +func (f *fakeRebuilder) rebuild(services []string) error { + if !atomic.CompareAndSwapInt32(&f.running, 0, 1) { + f.t.Errorf("rebuild() invoked while a previous rebuild was still in progress: services=%v", services) + } + defer atomic.StoreInt32(&f.running, 0) + + sorted := append([]string(nil), services...) + sort.Strings(sorted) + + f.mu.Lock() + f.calls = append(f.calls, sorted) + f.mu.Unlock() + + f.started <- sorted + return <-f.release +} + +func (f *fakeRebuilder) callCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.calls) +} + +// awaitStarted waits (with a bounded, generous timeout so a genuine bug +// hangs the test instead of the suite) for the next rebuild invocation to +// begin, returning the sorted services it was called with. +func awaitStarted(t *testing.T, f *fakeRebuilder) []string { + t.Helper() + select { + case services := <-f.started: + return services + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for rebuild to start") + return nil + } +} + +// assertNoRebuildStarts asserts that no new rebuild invocation begins +// within a short grace window. Since a correctly implemented scheduler +// decides synchronously (under its own lock) whether to start a rebuild or +// merely record it as pending, this is not a race against a slow producer: +// if the scheduler doesn't start a rebuild by the time this is called, it +// never will for the requests already issued. +func assertNoRebuildStarts(t *testing.T, f *fakeRebuilder) { + t.Helper() + select { + case services := <-f.started: + t.Fatalf("unexpected rebuild started with services=%v", services) + case <-time.After(50 * time.Millisecond): + // expected: nothing started + } +} + +// TestRebuildScheduler_FirstRequestStartsImmediately covers behavior (1): +// a request issued while idle must trigger a rebuild without being delayed +// or batched -- latency for an isolated edit must be unchanged. +func TestRebuildScheduler_FirstRequestStartsImmediately(t *testing.T) { + f := newFakeRebuilder(t) + scheduler := newRebuildScheduler(f.rebuild) + + scheduler.Request([]string{"web"}) + + services := awaitStarted(t, f) + assert.DeepEqual(t, services, []string{"web"}) + + f.release <- nil + scheduler.Wait() + + assert.Equal(t, f.callCount(), 1) +} + +// TestRebuildScheduler_RequestsDuringBuildAreCoalescedIntoPending covers +// behavior (2): requests arriving while a rebuild is already running must +// not start a new rebuild; they accumulate into a deduplicated pending set. +func TestRebuildScheduler_RequestsDuringBuildAreCoalescedIntoPending(t *testing.T) { + f := newFakeRebuilder(t) + scheduler := newRebuildScheduler(f.rebuild) + + scheduler.Request([]string{"web"}) + first := awaitStarted(t, f) + assert.DeepEqual(t, first, []string{"web"}) + + // These arrive while the first rebuild is still in flight (we haven't + // released it yet) and must not trigger immediate rebuilds. + scheduler.Request([]string{"api"}) + scheduler.Request([]string{"web"}) // duplicate, must not double up + scheduler.Request([]string{"api", "worker"}) + + assertNoRebuildStarts(t, f) + assert.Equal(t, f.callCount(), 1) + + f.release <- nil + + // The pending requests above must still produce exactly one trailing + // rebuild (verified by TestRebuildScheduler_TrailingRebuildConsolidatesPendingServices); + // drain it here so Wait() isn't left blocking on it forever. + awaitStarted(t, f) + f.release <- nil + scheduler.Wait() +} + +// TestRebuildScheduler_TrailingRebuildConsolidatesPendingServices covers +// behavior (3): once the in-progress rebuild completes, exactly one +// trailing rebuild starts automatically, covering the union of every +// service requested while the first rebuild was running, and the pending +// set is cleared. +func TestRebuildScheduler_TrailingRebuildConsolidatesPendingServices(t *testing.T) { + f := newFakeRebuilder(t) + scheduler := newRebuildScheduler(f.rebuild) + + scheduler.Request([]string{"web"}) + awaitStarted(t, f) + + scheduler.Request([]string{"api"}) + scheduler.Request([]string{"web"}) + scheduler.Request([]string{"api", "worker"}) + + f.release <- nil // let the first rebuild finish + + trailing := awaitStarted(t, f) + assert.DeepEqual(t, trailing, []string{"api", "web", "worker"}) + + f.release <- nil + scheduler.Wait() + + // Exactly one trailing rebuild: nothing else was requested during it. + assert.Equal(t, f.callCount(), 2) +} + +// TestRebuildScheduler_TrailingRebuildsChainUntilPendingEmpty covers +// behavior (4): if new requests arrive while a trailing rebuild is +// running, another trailing rebuild fires when it completes, and this +// repeats until a rebuild finishes with nothing pending, at which point the +// scheduler goes idle. +func TestRebuildScheduler_TrailingRebuildsChainUntilPendingEmpty(t *testing.T) { + f := newFakeRebuilder(t) + scheduler := newRebuildScheduler(f.rebuild) + + scheduler.Request([]string{"web"}) + assert.DeepEqual(t, awaitStarted(t, f), []string{"web"}) + + scheduler.Request([]string{"api"}) + f.release <- nil // first rebuild done, "api" is pending -> triggers 2nd rebuild + + assert.DeepEqual(t, awaitStarted(t, f), []string{"api"}) + + // A request arrives *during* the trailing rebuild: must chain into a 3rd. + scheduler.Request([]string{"db"}) + f.release <- nil // 2nd rebuild done, "db" is pending -> triggers 3rd rebuild + + assert.DeepEqual(t, awaitStarted(t, f), []string{"db"}) + + // Nothing requested during the 3rd rebuild: scheduler must return to idle. + f.release <- nil + scheduler.Wait() + + assertNoRebuildStarts(t, f) + assert.Equal(t, f.callCount(), 3) +} + +// TestRebuildScheduler_RebuildErrorDoesNotBlockPendingProcessing covers +// behavior (6): a failed rebuild must not prevent a subsequently pending +// rebuild from being processed. +func TestRebuildScheduler_RebuildErrorDoesNotBlockPendingProcessing(t *testing.T) { + f := newFakeRebuilder(t) + scheduler := newRebuildScheduler(f.rebuild) + + scheduler.Request([]string{"web"}) + awaitStarted(t, f) + + scheduler.Request([]string{"api"}) + f.release <- errors.New("build failed") // first rebuild fails + + // Despite the error, the pending "api" request must still be processed. + trailing := awaitStarted(t, f) + assert.DeepEqual(t, trailing, []string{"api"}) + + f.release <- nil + scheduler.Wait() + + assert.Equal(t, f.callCount(), 2) +} + +// TestRebuildScheduler_NoConcurrentRebuilds covers behavior (5): at most +// one rebuild must ever be running at a time, and the scheduler's internal +// state (pending set, "building" flag) must not race under concurrent +// requests. Run with `go test -race` to make both properties meaningful. +func TestRebuildScheduler_NoConcurrentRebuilds(t *testing.T) { + var running int32 + var maxObservedConcurrency int32 + + rebuild := func(_ []string) error { + n := atomic.AddInt32(&running, 1) + defer atomic.AddInt32(&running, -1) + + for { + max := atomic.LoadInt32(&maxObservedConcurrency) + if n <= max || atomic.CompareAndSwapInt32(&maxObservedConcurrency, max, n) { + break + } + } + + // Widen the window during which a concurrency bug would be + // observable. This does not make the test's pass/fail outcome + // depend on timing: it only increases the odds of *catching* a + // bug, it can never cause a false failure. + time.Sleep(2 * time.Millisecond) + return nil + } + + scheduler := newRebuildScheduler(rebuild) + + const goroutines = 50 + var wg sync.WaitGroup + wg.Add(goroutines) + for i := 0; i < goroutines; i++ { + go func() { + defer wg.Done() + scheduler.Request([]string{"svc"}) + }() + } + wg.Wait() + + scheduler.Wait() + + assert.Equal(t, atomic.LoadInt32(&maxObservedConcurrency), int32(1)) +} + +// TestRebuildScheduler_InFlightOrPending covers the query callers use to +// avoid racing a plain restart against a rebuild for the same service, +// whether that rebuild is currently running or only queued as trailing. +func TestRebuildScheduler_InFlightOrPending(t *testing.T) { + f := newFakeRebuilder(t) + scheduler := newRebuildScheduler(f.rebuild) + + assert.Equal(t, scheduler.InFlightOrPending("web"), false) + + scheduler.Request([]string{"web"}) + awaitStarted(t, f) + assert.Equal(t, scheduler.InFlightOrPending("web"), true) + assert.Equal(t, scheduler.InFlightOrPending("api"), false) + + // Queued for the trailing rebuild while "web" is still building. + scheduler.Request([]string{"api"}) + assert.Equal(t, scheduler.InFlightOrPending("api"), true) + + f.release <- nil // "web" finishes, "api" starts as the trailing rebuild + awaitStarted(t, f) + assert.Equal(t, scheduler.InFlightOrPending("api"), true) + assert.Equal(t, scheduler.InFlightOrPending("web"), false) + + f.release <- nil + scheduler.Wait() + + assert.Equal(t, scheduler.InFlightOrPending("api"), false) +} diff --git a/pkg/compose/watch.go b/pkg/compose/watch.go index baba7707d6..5d2197d818 100644 --- a/pkg/compose/watch.go +++ b/pkg/compose/watch.go @@ -373,11 +373,19 @@ func isSync(trigger types.Trigger) bool { func (s *composeService) watchEvents(ctx context.Context, project *types.Project, options api.WatchOptions, watcher watch.Notify, syncer sync.Syncer, rules []watchRule) error { ctx, cancel := context.WithCancel(ctx) - defer cancel() // debounce and group filesystem events so that we capture IDE saving many files as one "batch" event batchEvents := watch.BatchDebounceEvents(ctx, watcher.Events()) + scheduler := newRebuildScheduler(func(services []string) error { + return s.rebuild(ctx, project, services, options) + }) + // Rebuilds run asynchronously in their own goroutine(s); cancel first so + // an in-flight one can unwind, then wait for it to settle so none + // outlive this function. + defer scheduler.Wait() + defer cancel() + for { select { case <-ctx.Done(): @@ -406,7 +414,7 @@ func (s *composeService) watchEvents(ctx context.Context, project *types.Project } start := time.Now() logrus.Debugf("batch start: count[%d]", len(batch)) - err := s.handleWatchBatch(ctx, project, options, batch, rules, syncer) + err := s.handleWatchBatch(ctx, project, options, batch, rules, syncer, scheduler) if err != nil { logrus.Warnf("Error handling changed files: %v", err) // If context was canceled, exit immediately @@ -557,7 +565,9 @@ func (t tarDockerClient) Untar(ctx context.Context, id string, archive io.ReadCl return err } -func (s *composeService) handleWatchBatch(ctx context.Context, project *types.Project, options api.WatchOptions, batch []watch.FileEvent, rules []watchRule, syncer sync.Syncer) error { +func (s *composeService) handleWatchBatch(ctx context.Context, project *types.Project, options api.WatchOptions, + batch []watch.FileEvent, rules []watchRule, syncer sync.Syncer, scheduler *rebuildScheduler, +) error { var ( restart = map[string]bool{} syncfiles = map[string][]*sync.PathMapping{} @@ -590,15 +600,28 @@ func (s *composeService) handleWatchBatch(ctx context.Context, project *types.Pr } } - logrus.Debugf("watch actions: rebuild %d sync %d restart %d", len(rebuild), len(syncfiles), len(restart)) - if len(rebuild) > 0 { - err := s.rebuild(ctx, project, utils.MapKeys(rebuild), options) - if err != nil { - return err + scheduler.Request(utils.MapKeys(rebuild)) + } + + // A rebuild already recreates and starts the service; restarting one + // that has a rebuild in flight or queued -- from this batch or an + // earlier one still running asynchronously (see rebuildScheduler) -- + // would race the rebuild's create/start on the same container. + for service := range restart { + if scheduler.InFlightOrPending(service) { + logrus.Debugf("skipping restart for service %q: rebuild already in flight or pending", service) + delete(restart, service) } } + logrus.Debugf("watch actions: rebuild %d sync %d restart %d", len(rebuild), len(syncfiles), len(restart)) + + // A sync or exec below may still target a service whose container is + // being replaced by an in-flight asynchronous rebuild (from this batch or + // an earlier one): unlike restart, this is a conscious tradeoff, since + // syncer.Sync/exec resolve containers by lookup and simply fail loudly + // (rather than racing the container lifecycle) if the container is gone. for serviceName, pathMappings := range syncfiles { writeWatchSyncMessage(options.LogTo, serviceName, pathMappings) err := syncer.Sync(ctx, serviceName, pathMappings) diff --git a/pkg/compose/watch_test.go b/pkg/compose/watch_test.go index 963b5180b5..768a404c93 100644 --- a/pkg/compose/watch_test.go +++ b/pkg/compose/watch_test.go @@ -159,21 +159,64 @@ func TestWatch_Sync(t *testing.T) { }) assert.DeepEqual(t, expected, actual) - // Rebuild fails before sync actions from the same batch are processed. + // The rebuild triggered by "/rebuild" now runs asynchronously, so it no + // longer blocks the sync of "/sync/changed" from the same batch. + // synctest.Wait() only returns once the rebuild's goroutine has + // settled (it runs to completion here, exercising the mocked + // ImageList/ImageRemove prune calls), so the mock expectations above + // are already satisfied by the time we get here. watcher.Events() <- watch.NewFileEvent("/rebuild") watcher.Events() <- watch.NewFileEvent("/sync/changed") time.Sleep(watch.QuietPeriod) synctest.Wait() - select { - case batch := <-syncer.synced: - t.Fatalf("received unexpected events: %v", batch) - default: - // expected + actual = <-syncer.synced + expected = []*sync.PathMapping{ + {HostPath: "/sync/changed", ContainerPath: "/work/changed"}, } - // TODO: there's not a great way to assert that the rebuild attempt happened + assert.DeepEqual(t, expected, actual) }) } +// #14051 (follow-up): a rebuild running in the background from an earlier +// batch must not be raced by a plain restart of the same service triggered +// by a later batch -- restart must be skipped even though this batch itself +// carries no rebuild trigger. +func TestHandleWatchBatch_SkipsRestartForServiceWithRebuildInFlight(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + cli.EXPECT().Err().Return(streams.NewOut(os.Stderr)).AnyTimes() + // A correctly filtered restart never touches the Docker client at all. + cli.EXPECT().Client().Times(0) + service := composeService{dockerCli: cli} + + proj := types.Project{ + Name: "myProjectName", + Services: types.Services{ + "test": {Name: "test"}, + }, + } + + rules, err := getWatchRules(&types.DevelopConfig{ + Watch: []types.Trigger{ + {Path: "/restart", Action: "restart"}, + }, + }, types.ServiceConfig{Name: "test"}) + assert.NilError(t, err) + + release := make(chan error) + scheduler := newRebuildScheduler(func(_ []string) error { + return <-release + }) + scheduler.Request([]string{"test"}) // simulate a rebuild still running from an earlier batch + + err = service.handleWatchBatch(t.Context(), &proj, api.WatchOptions{LogTo: stdLogger{}}, + []watch.FileEvent{watch.NewFileEvent("/restart")}, rules, newFakeSyncer(), scheduler) + assert.NilError(t, err) + + release <- nil + scheduler.Wait() +} + type fakeSyncer struct { synced chan []*sync.PathMapping }