test+fix: red tests and fixes for upstream merge (stacked on #3)

.github/workflows/ci.yml

@@ -11,10 +11,13 @@ on:
   workflow_dispatch:
   schedule:
     - cron: '0 8 */6 * *' # every 6 days
-  pull_request:
+  push:
     branches:
-      - master
-      - main
+      - earthly-main
+      - gh_test_ci
+  # No branch filter, so PRs stacked on other PR branches
+  # (not just PRs into earthly-main) get CI too.
+  pull_request:
 
 jobs:
   validate:
@@ -28,13 +31,13 @@ jobs:
     steps:
       -
         name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
       -
         name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3.12.0
       -
         name: Run
-        uses: docker/bake-action@v4
+        uses: docker/bake-action@4a9a8d494466d37134e2bfca2d3a8de8fb2681ad # v5.13.0
         with:
           targets: ${{ matrix.target }}
 
@@ -48,13 +51,13 @@ jobs:
     steps:
       -
         name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
       -
         name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3.12.0
       -
         name: Run
-        uses: docker/bake-action@v4
+        uses: docker/bake-action@4a9a8d494466d37134e2bfca2d3a8de8fb2681ad # v5.13.0
         with:
           targets: ${{ matrix.target }}
 
@@ -73,18 +76,18 @@ jobs:
     steps:
       -
         name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
       -
         name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3.12.0
       -
         name: Test
-        uses: docker/bake-action@v4
+        uses: docker/bake-action@4a9a8d494466d37134e2bfca2d3a8de8fb2681ad # v5.13.0
         with:
           targets: test-${{ matrix.mode }}
       -
         name: Upload coverage
-        uses: codecov/codecov-action@v4
+        uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0
         with:
           directory: ${{ env.DESTDIR }}/coverage
           flags: unit,${{ matrix.mode }},go-${{ matrix.go }}
@@ -104,10 +107,10 @@ jobs:
     steps:
       -
         name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
       -
         name: Set up Go
-        uses: actions/setup-go@v5
+        uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5.6.0
         with:
           go-version: ${{ matrix.go }}
       -
@@ -118,7 +121,7 @@ jobs:
       -
         name: Upload coverage
         if: always()
-        uses: codecov/codecov-action@v4
+        uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0
         with:
           file: ./coverage.txt
           env_vars: RUNNER_OS
@@ -132,10 +135,10 @@ jobs:
     steps:
       -
         name: Checkout
-        uses: actions/checkout@v4
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0
       -
         name: Cache Vagrant boxes
-        uses: actions/cache@v4
+        uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
         with:
           path: ~/.vagrant.d/boxes
           key: ${{ runner.os }}-vagrant-${{ hashFiles('hack/Vagrantfile.freebsd13') }}
@@ -173,7 +176,7 @@ jobs:
       -
         name: Upload coverage
         if: always()
-        uses: codecov/codecov-action@v4
+        uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0
         with:
           file: ./coverage.txt
           flags: unit,freebsd

receive.go

@@ -221,10 +221,6 @@ func (r *receiver) run(ctx context.Context) error {
 					}
 					break
 				}
-				if r.verboseProgressCb != nil {
-					r.verboseProgressCb(p.Stat.Path, StatusStat, p.Size())
-				}
-
 				// normalize unix wire-specific paths to platform-specific paths
 				path := filepath.FromSlash(p.Stat.Path)
 				if filepath.ToSlash(path) != p.Stat.Path {
@@ -234,6 +230,12 @@ func (r *receiver) run(ctx context.Context) error {
 				p.Stat.Path = path
 				p.Stat.Linkname = filepath.FromSlash(p.Stat.Linkname)
 
+				// earthly-specific: report the native path, matching the
+				// StatusReceiving/StatusReceived paths (r.files keys) and NotifyHashed.
+				if r.verboseProgressCb != nil {
+					r.verboseProgressCb(path, StatusStat, p.Size())
+				}
+
 				if fileCanRequestData(os.FileMode(p.Stat.Mode)) {
 					r.mu.Lock()
 					r.files[p.Stat.Path] = i

send.go

@@ -114,9 +114,6 @@ func (s *sender) run(ctx context.Context) error {
 }
 
 func (s *sender) updateProgress(size int, last bool) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.progressCurrent += size
 	if s.progressCb != nil {
 		s.progressCurrentMu.Lock()
 		defer s.progressCurrentMu.Unlock()
@@ -149,7 +146,9 @@ func (s *sender) sendFile(h *sendHandle) error {
 			return err
 		}
 		if s.verboseProgressCb != nil {
-			s.verboseProgressCb(h.path, StatusSent, fs.bytesWritten)
+			// h.path is the unix wire path from s.files; report the native
+			// path, matching the StatusStat path reported by walk.
+			s.verboseProgressCb(filepath.FromSlash(h.path), StatusSent, fs.bytesWritten)
 		}
 	}
 	return s.conn.SendMsg(&types.Packet{ID: h.id, Type: types.PACKET_DATA})

send_progress_test.go

@@ -0,0 +1,87 @@
+package fsutil
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/tonistiigi/fsutil/types"
+	"golang.org/x/sync/errgroup"
+)
+
+// progressCountingStream wraps a Stream and sums the Size() of every packet
+// that the sender is supposed to account for in its progress callback: each
+// non-terminal STAT packet and each non-empty DATA packet.
+type progressCountingStream struct {
+	Stream
+	mu    sync.Mutex
+	total int
+}
+
+func (c *progressCountingStream) SendMsg(m interface{}) error {
+	if p, ok := m.(*types.Packet); ok {
+		if (p.Type == types.PACKET_STAT && p.Stat != nil) || (p.Type == types.PACKET_DATA && len(p.Data) > 0) {
+			c.mu.Lock()
+			c.total += p.Size()
+			c.mu.Unlock()
+		}
+	}
+	return c.Stream.SendMsg(m)
+}
+
+// TestSendProgressTotal checks that the cumulative size reported to the
+// sender's progressCb matches the bytes the sender actually put on the wire.
+func TestSendProgressTotal(t *testing.T) {
+	src := t.TempDir()
+	files := map[string][]byte{
+		"small":     []byte("data1"),
+		"dir/empty": nil,
+		// Larger than the 32KiB copy buffer so it is split into several
+		// DATA packets.
+		"dir/large": bytes.Repeat([]byte("x"), 100*1024+7),
+	}
+	for name, dt := range files {
+		p := filepath.Join(src, filepath.FromSlash(name))
+		require.NoError(t, os.MkdirAll(filepath.Dir(p), 0o755))
+		require.NoError(t, os.WriteFile(p, dt, 0o644))
+	}
+
+	fs, err := NewFS(src)
+	require.NoError(t, err)
+
+	var (
+		mu        sync.Mutex
+		lastTotal = -1
+		lastCalls int
+	)
+	progressCb := func(total int, last bool) {
+		mu.Lock()
+		defer mu.Unlock()
+		if last {
+			lastCalls++
+			lastTotal = total
+		}
+	}
+
+	eg, ctx := errgroup.WithContext(context.Background())
+	s1, s2 := sockPairProto(ctx)
+	counting := &progressCountingStream{Stream: s1}
+
+	eg.Go(func() error {
+		defer s1.(*fakeConnProto).closeSend()
+		return Send(ctx, counting, fs, progressCb, nil)
+	})
+	eg.Go(func() error {
+		return Receive(ctx, s2, t.TempDir(), ReceiveOpt{})
+	})
+	require.NoError(t, eg.Wait())
+
+	want := counting.total
+	require.Positive(t, want)
+	require.Equal(t, 1, lastCalls, "progressCb should be called exactly once with last=true")
+	require.Equal(t, want, lastTotal, "final progress total should equal the size of STAT and DATA packets sent")
+}

verbose_progress_test.go

@@ -0,0 +1,128 @@
+package fsutil
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"sort"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+)
+
+// verboseRecorder collects the paths passed to a VerboseProgressCB, keyed by
+// status.
+type verboseRecorder struct {
+	mu    sync.Mutex
+	paths map[VerboseProgressStatus][]string
+}
+
+func newVerboseRecorder() *verboseRecorder {
+	return &verboseRecorder{paths: map[VerboseProgressStatus][]string{}}
+}
+
+func (r *verboseRecorder) cb(p string, status VerboseProgressStatus, _ int) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	for _, existing := range r.paths[status] {
+		if existing == p {
+			return
+		}
+	}
+	r.paths[status] = append(r.paths[status], p)
+}
+
+func (r *verboseRecorder) get(status VerboseProgressStatus) []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := append([]string(nil), r.paths[status]...)
+	sort.Strings(out)
+	return out
+}
+
+// TestVerboseProgressPathsConsistent checks that, on each side of a transfer,
+// the verbose progress callback reports a file under the same path for every
+// status, and that the receiver reports the same (platform-native) path that
+// it hands to NotifyHashed.
+//
+// On Unix filepath.FromSlash/ToSlash are no-ops so this passes. On Windows the
+// merge of upstream's wire-path normalisation leaves:
+//   - sender: StatusStat gets the native walk path (`dir\b`) while
+//     StatusSent gets the wire path stored in s.files ("dir/b");
+//   - receiver: StatusStat is called with p.Stat.Path before it is converted
+//     with filepath.FromSlash ("dir/b") while StatusReceiving/Received
+//     get the native r.files key via pipeNames (`dir\b`). It also fires
+//     before the "unrepresentable path" check, so a rejected path is still
+//     reported as stat'ed.
+//
+// This test cannot fail on macOS/Linux; it is red only on Windows.
+func TestVerboseProgressPathsConsistent(t *testing.T) {
+	src := t.TempDir()
+	names := []string{"a", "dir/b", "dir/sub/c"}
+	for _, name := range names {
+		p := filepath.Join(src, filepath.FromSlash(name))
+		require.NoError(t, os.MkdirAll(filepath.Dir(p), 0o755))
+		require.NoError(t, os.WriteFile(p, []byte("data-"+name), 0o644))
+	}
+	var wantFiles []string
+	for _, name := range names {
+		wantFiles = append(wantFiles, filepath.FromSlash(name))
+	}
+	sort.Strings(wantFiles)
+
+	fs, err := NewFS(src)
+	require.NoError(t, err)
+
+	sendRec := newVerboseRecorder()
+	recvRec := newVerboseRecorder()
+
+	var hashedMu sync.Mutex
+	var hashed []string
+	notifyHashed := func(kind ChangeKind, p string, fi os.FileInfo, err error) error {
+		if fi != nil && fi.Mode().IsRegular() {
+			hashedMu.Lock()
+			hashed = append(hashed, p)
+			hashedMu.Unlock()
+		}
+		return nil
+	}
+
+	eg, ctx := errgroup.WithContext(context.Background())
+	s1, s2 := sockPairProto(ctx)
+	eg.Go(func() error {
+		defer s1.(*fakeConnProto).closeSend()
+		return Send(ctx, s1, fs, nil, sendRec.cb)
+	})
+	eg.Go(func() error {
+		return Receive(ctx, s2, t.TempDir(), ReceiveOpt{
+			NotifyHashed:      notifyHashed,
+			ContentHasher:     simpleSHA256Hasher,
+			VerboseProgressCb: recvRec.cb,
+		})
+	})
+	require.NoError(t, eg.Wait())
+
+	sort.Strings(hashed)
+	require.Equal(t, wantFiles, hashed, "NotifyHashed paths")
+
+	// Sender: every file reported as sent must have been reported as stat'ed
+	// under the identical path.
+	sendStat := sendRec.get(StatusStat)
+	sent := sendRec.get(StatusSent)
+	require.Len(t, sent, len(names), "sender StatusSent paths: %v", sent)
+	for _, p := range sent {
+		require.Contains(t, sendStat, p, "sender StatusSent path %q was not reported with StatusStat", p)
+	}
+
+	// Receiver: the same, and they must match what NotifyHashed reported.
+	recvStat := recvRec.get(StatusStat)
+	received := recvRec.get(StatusReceived)
+	receiving := recvRec.get(StatusReceiving)
+	require.Equal(t, wantFiles, received, "receiver StatusReceived paths")
+	require.Equal(t, wantFiles, receiving, "receiver StatusReceiving paths")
+	for _, p := range received {
+		require.Contains(t, recvStat, p, "receiver StatusReceived path %q was not reported with StatusStat", p)
+	}
+}