diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 283c4d4..fb6598c 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/receive.go b/receive.go index afb118b..dc73351 100644 --- a/receive.go +++ b/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 diff --git a/send.go b/send.go index 8e69422..908ba64 100644 --- a/send.go +++ b/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}) diff --git a/send_progress_test.go b/send_progress_test.go new file mode 100644 index 0000000..fb57303 --- /dev/null +++ b/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") +} diff --git a/verbose_progress_test.go b/verbose_progress_test.go new file mode 100644 index 0000000..537274c --- /dev/null +++ b/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) + } +}