Avoid concurrent wrapper generation

We had logic to debounce regenerating wrappers,
but if there were a large stream of FS events it could
still lead to several instances of the code generation
running concurrently. Fix this by tracking when the
last run completed, and only then re-running the
code generation.

cli/daemon/apps/apps.go

@@ -393,19 +393,24 @@ func (i *Instance) beginWatch() error {
 		}
 
 		go func() {
-			for range i.watcher.EventsReady {
-				batch := i.watcher.GetEventsBatch()
-				events := batch.Events()
-
-				if i.mgr != nil {
-					i.mgr.onWatchEvent(i, events)
-				}
-
-				i.watchMu.Lock()
-				watchers := i.watchers
-				i.watchMu.Unlock()
-				for _, sub := range watchers {
-					sub.f(i, events)
+			for {
+				select {
+				case <-i.watcher.Done():
+					return
+				case <-i.watcher.EventsReady:
+					batch := i.watcher.GetEventsBatch()
+					events := batch.Events()
+
+					if i.mgr != nil {
+						i.mgr.onWatchEvent(i, events)
+					}
+
+					i.watchMu.Lock()
+					watchers := i.watchers
+					i.watchMu.Unlock()
+					for _, sub := range watchers {
+						sub.f(i, events)
+					}
 				}
 			}
 		}()

cli/daemon/daemon.go

@@ -52,7 +52,7 @@ type Server struct {
 	availableVer     atomic.Value // string
 
 	appDebounceMu sync.Mutex
-	appDebouncers map[*apps.Instance]debouncer
+	appDebouncers map[*apps.Instance]*regenerateCodeDebouncer
 
 	daemonpb.UnimplementedDaemonServer
 }
@@ -67,7 +67,7 @@ func New(appsMgr *apps.Manager, mgr *run.Manager, cm *sqldb.ClusterManager, sm *
 		ns:      ns,
 		streams: make(map[string]*streamLog),
 
-		appDebouncers: make(map[*apps.Instance]debouncer),
+		appDebouncers: make(map[*apps.Instance]*regenerateCodeDebouncer),
 	}
 
 	mgr.AddListener(srv)

cli/daemon/watch.go

@@ -7,6 +7,7 @@ import (
 	"io/fs"
 	"os"
 	"path/filepath"
+	"sync"
 	"time"
 
 	"github.com/bep/debounce"
@@ -44,12 +45,52 @@ func (s *Server) onWatchEvent(i *apps.Instance, events []watcher.Event) {
 	s.appDebounceMu.Lock()
 	deb := s.appDebouncers[i]
 	if deb == nil {
-		deb = debounce.New(100 * time.Millisecond)
+		deb = &regenerateCodeDebouncer{
+			debounce: debounce.New(100 * time.Millisecond),
+			doRun:    func() { s.regenerateUserCode(context.Background(), i) },
+		}
 		s.appDebouncers[i] = deb
 	}
 	s.appDebounceMu.Unlock()
 
-	deb(func() { s.regenerateUserCode(context.Background(), i) })
+	deb.ChangeEvent()
+}
+
+type regenerateCodeDebouncer struct {
+	debounce func(func())
+	mu       sync.Mutex
+	running  bool
+	runAfter bool
+
+	doRun func()
+}
+
+func (g *regenerateCodeDebouncer) ChangeEvent() {
+	g.debounce(func() {
+		g.mu.Lock()
+
+		// If we're already running, mark to run again when complete.
+		if g.running {
+			g.runAfter = true
+			g.mu.Unlock()
+			return
+		}
+
+		// Otherwise, keep re-running for as long as change events come in.
+		g.running = true
+		g.runAfter = true // to start us off, at least once.
+		for g.runAfter {
+			g.runAfter = false // reset for next time
+			g.mu.Unlock()
+			g.doRun() // actually run
+			g.mu.Lock()
+		}
+
+		// If we get here g.runAfter nobody requested another run, so we can stop.
+		g.running = false
+
+		g.mu.Unlock()
+	})
 }
 
 func (s *Server) regenerateUserCode(ctx context.Context, app *apps.Instance) {
@@ -99,5 +140,3 @@ func (s *Server) updateGitIgnore(i *apps.Instance) error {
 	}
 	return nil
 }
-
-type debouncer = func(fn func())

pkg/watcher/watcher.go

@@ -177,8 +177,10 @@ func (w *Watcher) GetEventsBatch() *Events {
 }
 
 func (w *Watcher) Close() error {
-	w.stop <- struct{}{}
-	close(w.EventsReady)
 	close(w.stop)
 	return nil
 }
+
+func (w *Watcher) Done() <-chan struct{} {
+	return w.stop
+}