Merge master to resolve conflicts with folbricht/desync#323

README.md

@@ -12,7 +12,7 @@ It implements the [casync](https://github.com/systemd/casync) format and interop
 
 ## Where it fits
 
-- **Updating devices in the field.** An appliance with A/B partitions seeds from the partition it is running and writes the new image straight to the other one, downloading only the chunks that aren't already on disk. An update interrupted by a dropped connection restarts without fetching the completed chunks again.
+- **Updating devices in the field.** An appliance with A/B partitions seeds from the partition it is running and writes the new image straight to the other one, downloading only the chunks that aren't already on disk. An update interrupted by a dropped connection restarts without fetching the completed chunks again. A device with a single copy of the image can be updated in place instead, rearranging the data already on it and downloading only the chunks it lacks.
 - **Serving a fleet from one copy.** Each client works out for itself which chunks it is missing, so one set of static files serves every client, whichever version it is starting from. Hash-named chunks never change, so a CDN can cache them indefinitely, and a `chunk-server` with a local cache at each site means a chunk crosses the WAN once rather than once per machine.
 - **Using an image before it has downloaded.** `mount-index` exposes an index as a file over FUSE and fetches chunks as they are read, so a VM can boot from a disk image in a remote store. With `--cor-file` it fills a local sparse copy on the way.
 - **Keeping every build.** Publish each build to the same store and it grows by what changed, not by another full image. `prune` removes the chunks only old versions still reference.
@@ -39,7 +39,7 @@ How much you save depends on the data: on how much changed between versions, and
 
 | | desync | casync | rsync | zsync | OCI / ORAS |
 | --- | --- | --- | --- | --- | --- |
-| Reuses local data as a seed | yes | yes | yes, the destination file | yes | no |
+| Reuses local data as a seed | yes, including the destination file | yes | yes, the destination file | yes | no |
 | Server-side work per client | none | none | delta computed per transfer | none | none |
 | Server requirement | any static file host | any static file host | rsync daemon or SSH | static host with range requests | registry |
 | Deduplication across versions in the store | yes | yes | no | no | whole layers only |

assemble-backfill.go

@@ -0,0 +1,118 @@
+package desync
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"sync"
+	"time"
+)
+
+// seedFileState is what a seed file looked like when its chunks were
+// validated. Comparing it with the file after assembly shows whether the file
+// was written to while it was read.
+type seedFileState struct {
+	info os.FileInfo
+
+	// recent is set when the file was modified shortly before it was
+	// validated. Some filesystems keep modification times to the second or
+	// worse, a later write could then leave the time unchanged.
+	recent bool
+}
+
+// seedTimeResolution is the coarsest resolution of modification times among
+// common filesystems, FAT's two seconds.
+const seedTimeResolution = 2 * time.Second
+
+func newSeedFileState(info os.FileInfo) seedFileState {
+	return seedFileState{
+		info:   info,
+		recent: time.Since(info.ModTime()) < seedTimeResolution,
+	}
+}
+
+// changed reports whether the file may have been written to since its state
+// was recorded. A block device has no meaningful modification time, it always
+// counts as changed.
+func (s seedFileState) changed(name string) bool {
+	if s.recent || isDevice(s.info.Mode()) {
+		return true
+	}
+	info, err := os.Stat(name)
+	if err != nil {
+		return true
+	}
+	return !os.SameFile(info, s.info) ||
+		info.Size() != s.info.Size() ||
+		!info.ModTime().Equal(s.info.ModTime())
+}
+
+// chunksInTarget reports for every chunk of the index whether the file holds
+// its data at the chunk's position, hashing them with n goroutines. Null
+// chunks are compared with zeros instead, which is much cheaper than hashing
+// in images that are mostly empty space. The progress is counted in chunks.
+func chunksInTarget(name string, idx Index, n int, pb ProgressBar) ([]bool, error) {
+	f, err := os.Open(name)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+
+	pb.SetTotal(len(idx.Chunks))
+	pb.Start()
+	defer pb.Finish()
+
+	null := NewNullChunk(idx.Index.ChunkSizeMax)
+	found := make([]bool, len(idx.Chunks))
+	var wg sync.WaitGroup
+	work := make(chan int)
+	for range max(n, 1) {
+		wg.Go(func() {
+			buf := make([]byte, idx.Index.ChunkSizeMax)
+			for i := range work {
+				c := idx.Chunks[i]
+				if c.ID != null.ID {
+					found[i] = chunkInPlace(f, c, buf)
+				} else {
+					b := buf[:c.Size]
+					_, err := f.ReadAt(b, int64(c.Start))
+					found[i] = err == nil && bytes.Equal(b, null.Data)
+				}
+				pb.Increment()
+			}
+		})
+	}
+	for i := range idx.Chunks {
+		work <- i
+	}
+	close(work)
+	wg.Wait()
+	return found, nil
+}
+
+// backfill checks every chunk of the assembled file and writes the ones that
+// don't match from the store. It's used when a seed changed while it was read,
+// whatever came from it may be damaged, including what the self-seed copied
+// from there on.
+func backfill(name string, idx Index, s Store, n int, stats *ExtractStats) error {
+	found, err := chunksInTarget(name, idx, n, NewProgressBar("Verifying "))
+	if err != nil {
+		return err
+	}
+	f, err := os.OpenFile(name, os.O_RDWR, 0)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	for i, ok := range found {
+		if ok {
+			continue
+		}
+		c := idx.Chunks[i]
+		if _, _, err := (&copyFromStore{store: s, chunk: c}).Execute(f); err != nil {
+			return fmt.Errorf("chunk %s at offset %d came from a seed that changed during assembly: %w", c.ID, c.Start, err)
+		}
+		stats.addChunksBackfilled(1)
+	}
+	return nil
+}

assemble-backfill_test.go

@@ -0,0 +1,190 @@
+package desync
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// writeOldFile writes a file whose modification time lies well in the past,
+// so only a later write can make it look changed.
+func writeOldFile(t *testing.T, name string, data []byte) {
+	t.Helper()
+	require.NoError(t, os.WriteFile(name, data, 0644))
+	past := time.Now().Add(-time.Hour)
+	require.NoError(t, os.Chtimes(name, past, past))
+}
+
+func TestSeedFileStateChanged(t *testing.T) {
+	// Record the file the way validation does, from the open file. On
+	// Windows, the identity of a file stat'ed by name is looked up by name
+	// again when it's compared, which would find the replacement.
+	record := func(t *testing.T, name string) seedFileState {
+		t.Helper()
+		f, err := os.Open(name)
+		require.NoError(t, err)
+		defer f.Close()
+		info, err := f.Stat()
+		require.NoError(t, err)
+		return newSeedFileState(info)
+	}
+	dir := t.TempDir()
+
+	t.Run("unchanged", func(t *testing.T) {
+		name := filepath.Join(dir, "unchanged")
+		writeOldFile(t, name, []byte("seed data"))
+		assert.False(t, record(t, name).changed(name))
+	})
+	t.Run("written to", func(t *testing.T) {
+		name := filepath.Join(dir, "written")
+		writeOldFile(t, name, []byte("seed data"))
+		state := record(t, name)
+		require.NoError(t, os.WriteFile(name, []byte("SEED DATA"), 0644))
+		assert.True(t, state.changed(name))
+	})
+	t.Run("replaced", func(t *testing.T) {
+		name := filepath.Join(dir, "replaced")
+		writeOldFile(t, name, []byte("seed data"))
+		state := record(t, name)
+		other := filepath.Join(dir, "other")
+		writeOldFile(t, other, []byte("seed data"))
+		require.NoError(t, os.Chtimes(other, state.info.ModTime(), state.info.ModTime()))
+		require.NoError(t, os.Rename(other, name))
+		assert.True(t, state.changed(name), "same size and time, different file")
+	})
+	t.Run("removed", func(t *testing.T) {
+		name := filepath.Join(dir, "removed")
+		writeOldFile(t, name, []byte("seed data"))
+		state := record(t, name)
+		require.NoError(t, os.Remove(name))
+		assert.True(t, state.changed(name))
+	})
+	t.Run("modified just before", func(t *testing.T) {
+		// A write within the time resolution of the filesystem might not
+		// change the modification time
+		name := filepath.Join(dir, "recent")
+		require.NoError(t, os.WriteFile(name, []byte("seed data"), 0644))
+		assert.True(t, record(t, name).changed(name))
+	})
+}
+
+// hookStore calls a function before handing out a chunk.
+type hookStore struct {
+	Store
+	hook func()
+}
+
+func (s hookStore) GetChunk(id ChunkID) (*Chunk, error) {
+	s.hook()
+	return s.Store.GetChunk(id)
+}
+
+// A seed that's written to during assembly is copied into the target in its
+// new state. The damaged chunks are found after assembly and taken from the
+// store instead.
+func TestAssembleSeedChangedDuringAssembly(t *testing.T) {
+	chunks := randomChunks(1024, 768)
+	fromStore, fromSeed := chunks[0], chunks[1]
+
+	// The store chunk comes first. With one worker it's written before the
+	// seed chunk, and fetching it overwrites the seed with other data of the
+	// same size.
+	setup := func(t *testing.T, store Store) (string, *ExtractStats, error) {
+		t.Helper()
+		dir := t.TempDir()
+		seedFile := filepath.Join(dir, "seed")
+		writeOldFile(t, seedFile, fromSeed.data)
+		seed, err := NewFileSeed(filepath.Join(dir, "target"), seedFile, chunkIndex(fromSeed), IndexRegenerationSettings{})
+		require.NoError(t, err)
+
+		// The hook runs on a worker, it can't stop the test
+		var once sync.Once
+		hooked := hookStore{Store: store, hook: func() {
+			once.Do(func() {
+				assert.NoError(t, os.WriteFile(seedFile, make([]byte, len(fromSeed.data)), 0644))
+			})
+		}}
+		target := filepath.Join(dir, "target")
+		stats, err := AssembleFile(context.Background(), target, chunkIndex(fromStore, fromSeed), hooked,
+			[]Seed{seed}, AssembleOptions{N: 1, InvalidSeedAction: InvalidSeedActionBailOut})
+		return target, stats, err
+	}
+
+	t.Run("backfilled from the store", func(t *testing.T) {
+		target, stats, err := setup(t, chunkStore(fromStore, fromSeed))
+		require.NoError(t, err)
+		assert.Equal(t, uint64(1), stats.ChunksFromSeeds)
+		assert.Equal(t, uint64(1), stats.ChunksBackfilled)
+
+		got, err := os.ReadFile(target)
+		require.NoError(t, err)
+		require.Equal(t, chunkContent(fromStore, fromSeed), got)
+	})
+
+	t.Run("not in the store", func(t *testing.T) {
+		_, _, err := setup(t, chunkStore(fromStore))
+		require.ErrorContains(t, err, "came from a seed that changed during assembly")
+	})
+}
+
+// Without a change, the output isn't checked again.
+func TestAssembleSeedUnchanged(t *testing.T) {
+	chunks := randomChunks(1024, 768)
+	fromStore, fromSeed := chunks[0], chunks[1]
+	dir := t.TempDir()
+	seedFile := filepath.Join(dir, "seed")
+	writeOldFile(t, seedFile, fromSeed.data)
+	target := filepath.Join(dir, "target")
+	seed, err := NewFileSeed(target, seedFile, chunkIndex(fromSeed), IndexRegenerationSettings{})
+	require.NoError(t, err)
+
+	// Checking the output would fetch nothing, so count the chunks the
+	// store hands out instead: only the one assembly needs.
+	var fetched atomic.Int32
+	store := hookStore{Store: chunkStore(fromStore, fromSeed), hook: func() { fetched.Add(1) }}
+	stats, err := AssembleFile(context.Background(), target, chunkIndex(fromStore, fromSeed), store,
+		[]Seed{seed}, testAssembleOptions)
+	require.NoError(t, err)
+	assert.Equal(t, uint64(0), stats.ChunksBackfilled)
+	assert.Equal(t, int32(1), fetched.Load())
+}
+
+func TestChunksInTargetNullChunks(t *testing.T) {
+	data := randomChunks(64)[0]
+	null := testChunk{id: NewNullChunk(64).ID, data: make([]byte, 64)}
+	idx := chunkIndex(null, data, null)
+
+	tests := map[string]struct {
+		content  []byte
+		expected []bool
+	}{
+		"zeros in place": {
+			content:  chunkContent(null, data, null),
+			expected: []bool{true, true, true},
+		},
+		"one byte set": {
+			content:  join(make([]byte, 63), []byte{1}, data.data, make([]byte, 64)),
+			expected: []bool{false, true, true},
+		},
+		"file too short": {
+			content:  chunkContent(null, data, null)[:150],
+			expected: []bool{true, true, false},
+		},
+	}
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			name := filepath.Join(t.TempDir(), "target")
+			require.NoError(t, os.WriteFile(name, test.content, 0644))
+			found, err := chunksInTarget(name, idx, 2, NullProgressBar{})
+			require.NoError(t, err)
+			require.Equal(t, test.expected, found)
+		})
+	}
+}