initial support for `sync`

Signed-off-by: Nicolas De Loof <nicolas.deloof@gmail.com>

go.mod

@@ -16,7 +16,6 @@ require (
 	github.com/docker/docker v20.10.20+incompatible // replaced; see replace rule for actual version
 	github.com/docker/go-connections v0.4.0
 	github.com/docker/go-units v0.5.0
-	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/golang/mock v1.6.0
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-version v1.6.0

go.sum

@@ -206,9 +206,8 @@ github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoD
 github.com/form3tech-oss/jwt-go v3.2.3+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
-github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
-github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
 github.com/fvbommel/sortorder v1.0.2 h1:mV4o8B2hKboCdkJm+a7uX/SIpZob4JzUpc5GGnM45eo=
 github.com/fvbommel/sortorder v1.0.2/go.mod h1:uk88iVf1ovNn1iLfgUVU2F9o5eO30ui720w+kxuqRs0=
 github.com/getsentry/raven-go v0.0.0-20180121060056-563b81fc02b7/go.mod h1:KungGk8q33+aIAZUIVWZDr2OfAEBsO49PX4NzFV5kcQ=
@@ -910,7 +909,6 @@ golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220422013727-9388b58f7150/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.4.0 h1:Zr2JFtRQNX3BCZ8YtxRE9hNJYC8J6I1MVbMg6owUp18=
 golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=

pkg/compose/convergence.go

@@ -551,7 +551,7 @@ func (s *composeService) createMobyContainer(ctx context.Context, project *types
 }
 
 // getLinks mimics V1 compose/service.py::Service::_get_links()
-func (s composeService) getLinks(ctx context.Context, projectName string, service types.ServiceConfig, number int) ([]string, error) {
+func (s *composeService) getLinks(ctx context.Context, projectName string, service types.ServiceConfig, number int) ([]string, error) {
 	var links []string
 	format := func(k, v string) string {
 		return fmt.Sprintf("%s:%s", k, v)

pkg/compose/watch.go

@@ -17,7 +17,7 @@ package compose
 import (
 	"context"
 	"fmt"
-	"log"
+	"path/filepath"
 	"strings"
 	"time"
 
@@ -32,56 +32,29 @@ import (
 )
 
 type DevelopmentConfig struct {
+	Sync     map[string]string `json:"sync,omitempty"`
+	Excludes []string          `json:"excludes,omitempty"`
 }
 
 const quietPeriod = 2 * time.Second
 
 func (s *composeService) Watch(ctx context.Context, project *types.Project, services []string, options api.WatchOptions) error {
-	fmt.Fprintln(s.stderr(), "not implemented yet")
+	needRebuild := make(chan string)
+	needSync := make(chan api.CopyOptions, 5)
 
 	eg, ctx := errgroup.WithContext(ctx)
-	needRefresh := make(chan string)
 	eg.Go(func() error {
 		clock := clockwork.NewRealClock()
-		debounce(ctx, clock, quietPeriod, needRefresh, func(services []string) {
+		debounce(ctx, clock, quietPeriod, needRebuild, s.makeRebuildFn(ctx, project))
-			fmt.Fprintf(s.stderr(), "Updating %s after changes were detected\n", strings.Join(services, ", "))
-			imageIds, err := s.build(ctx, project, api.BuildOptions{
-				Services: services,
-			})
-			if err != nil {
-				fmt.Fprintf(s.stderr(), "Build failed")
-			}
-			for i, service := range project.Services {
-				if id, ok := imageIds[service.Name]; ok {
-					service.Image = id
-				}
-				project.Services[i] = service
-			}
-
-			err = s.Up(ctx, project, api.UpOptions{
-				Create: api.CreateOptions{
-					Services: services,
-					Inherit:  true,
-				},
-				Start: api.StartOptions{
-					Services: services,
-					Project:  project,
-				},
-			})
-			if err != nil {
-				fmt.Fprintf(s.stderr(), "Application failed to start after update")
-			}
-		})
 		return nil
 	})
 
+	eg.Go(s.makeSyncFn(ctx, project, needSync))
+
 	err := project.WithServices(services, func(service types.ServiceConfig) error {
-		var config DevelopmentConfig
+		config, err := loadDevelopmentConfig(service, project)
-		if y, ok := service.Extensions["x-develop"]; ok {
+		if err != nil {
-			err := mapstructure.Decode(y, &config)
+			return err
-			if err != nil {
-				return err
-			}
 		}
 		if service.Build == nil {
 			return errors.New("can't watch a service without a build section")
@@ -98,7 +71,7 @@ func (s *composeService) Watch(ctx context.Context, project *types.Project, serv
 			return err
 		}
 
-		fmt.Println("watching " + context)
+		fmt.Fprintf(s.stderr(), "watching %s\n", context)
 		err = watcher.Start()
 		if err != nil {
 			return err
@@ -106,13 +79,32 @@ func (s *composeService) Watch(ctx context.Context, project *types.Project, serv
 
 		eg.Go(func() error {
 			defer watcher.Close() //nolint:errcheck
+		WATCH:
 			for {
 				select {
 				case <-ctx.Done():
 					return nil
 				case event := <-watcher.Events():
-					log.Println("fs event :", event.Path())
+					fmt.Fprintf(s.stderr(), "change detected on %s\n", event.Path())
-					needRefresh <- service.Name
+
+					for src, dest := range config.Sync {
+						path := filepath.Clean(event.Path())
+						src = filepath.Clean(src)
+						if watch.IsChild(path, src) {
+							rel, err := filepath.Rel(src, path)
+							if err != nil {
+								return err
+							}
+							dest = filepath.Join(dest, rel)
+							needSync <- api.CopyOptions{
+								Source:      path,
+								Destination: fmt.Sprintf("%s:%s", service.Name, dest),
+							}
+							continue WATCH
+						}
+					}
+
+					needRebuild <- service.Name
 				case err := <-watcher.Errors():
 					return err
 				}
@@ -127,6 +119,73 @@ func (s *composeService) Watch(ctx context.Context, project *types.Project, serv
 	return eg.Wait()
 }
 
+func loadDevelopmentConfig(service types.ServiceConfig, project *types.Project) (DevelopmentConfig, error) {
+	var config DevelopmentConfig
+	if y, ok := service.Extensions["x-develop"]; ok {
+		err := mapstructure.Decode(y, &config)
+		if err != nil {
+			return DevelopmentConfig{}, err
+		}
+		for src, dest := range config.Sync {
+			if !filepath.IsAbs(src) {
+				delete(config.Sync, src)
+				src = filepath.Join(project.WorkingDir, src)
+				config.Sync[src] = dest
+			}
+		}
+	}
+	return config, nil
+}
+
+func (s *composeService) makeRebuildFn(ctx context.Context, project *types.Project) func(services []string) {
+	return func(services []string) {
+		fmt.Fprintf(s.stderr(), "Updating %s after changes were detected\n", strings.Join(services, ", "))
+		imageIds, err := s.build(ctx, project, api.BuildOptions{
+			Services: services,
+		})
+		if err != nil {
+			fmt.Fprintf(s.stderr(), "Build failed")
+		}
+		for i, service := range project.Services {
+			if id, ok := imageIds[service.Name]; ok {
+				service.Image = id
+			}
+			project.Services[i] = service
+		}
+
+		err = s.Up(ctx, project, api.UpOptions{
+			Create: api.CreateOptions{
+				Services: services,
+				Inherit:  true,
+			},
+			Start: api.StartOptions{
+				Services: services,
+				Project:  project,
+			},
+		})
+		if err != nil {
+			fmt.Fprintf(s.stderr(), "Application failed to start after update")
+		}
+	}
+}
+
+func (s *composeService) makeSyncFn(ctx context.Context, project *types.Project, needSync chan api.CopyOptions) func() error {
+	return func() error {
+		for {
+			select {
+			case <-ctx.Done():
+				return nil
+			case opt := <-needSync:
+				err := s.Copy(ctx, project.Name, opt)
+				if err != nil {
+					return err
+				}
+				fmt.Fprintf(s.stderr(), "%s updated\n", opt.Source)
+			}
+		}
+	}
+}
+
 func debounce(ctx context.Context, clock clockwork.Clock, delay time.Duration, input chan string, fn func(services []string)) {
 	services := utils.Set[string]{}
 	t := clock.AfterFunc(delay, func() {

pkg/watch/notify.go

@@ -23,7 +23,9 @@ import (
 	"path/filepath"
 	"runtime"
 	"strconv"
-	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/tilt-dev/fsnotify"
 )
 
 var (
@@ -86,7 +88,7 @@ func NewWatcher(paths []string, ignore PathMatcher) (Notify, error) {
 	return newWatcher(paths, ignore)
 }
 
-const WindowsBufferSizeEnvVar = "TILT_WATCH_WINDOWS_BUFFER_SIZE"
+const WindowsBufferSizeEnvVar = "COMPOSE_WATCH_WINDOWS_BUFFER_SIZE"
 
 const defaultBufferSize int = 65536
 
@@ -102,5 +104,5 @@ func DesiredWindowsBufferSize() int {
 }
 
 func IsWindowsShortReadError(err error) bool {
-	return runtime.GOOS == "windows" && err != nil && strings.Contains(err.Error(), "short read")
+	return runtime.GOOS == "windows" && !errors.Is(err, fsnotify.ErrEventOverflow)
 }