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 <[email protected]>
2026-10-12 02:13:08 +00:00 · 2026-09-14 10:47:07 +02:00
parent 8f1a46c871
commit 53188a4896
4 changed files with 495 additions and 15 deletions
@@ -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()
+}
@@ -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)
+}
@@ -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)
@@ -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
 }