Fixed empty object IDs in checkpoints (#649)

* object: fixed race condition between Result() and Checkpoint()

This would sometimes result in indirect objects having empty object IDs.

Fixes #648

* upload: ensure checkpoints never containt empty object IDs.

* testing: reduce armhf test weight

internal/buf/pool_test.go

@@ -22,14 +22,22 @@ func TestPool(t *testing.T) {
 
 	runtime.ReadMemStats(&ms1)
 
+	repeat := 1000000
+	numGoRoutines := 30
+
+	if runtime.GOARCH != "amd64" {
+		repeat = 10000
+		numGoRoutines = 10
+	}
+
 	// 30 gorouties, each allocating and releasing memory 1 M times
-	for i := 0; i < 30; i++ {
+	for i := 0; i < numGoRoutines; i++ {
 		wg.Add(1)
 
 		go func() {
 			defer wg.Done()
 
-			for j := 0; j < 1000000; j++ {
+			for j := 0; j < repeat; j++ {
 				const allocSize = 100000
 
 				b := a.Allocate(allocSize)

repo/object/object_manager_test.go

@@ -17,7 +17,9 @@ import (
 	"testing"
 
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 
+	"github.com/kopia/kopia/internal/clock"
 	"github.com/kopia/kopia/internal/testlogging"
 	"github.com/kopia/kopia/repo/blob"
 	"github.com/kopia/kopia/repo/compression"
@@ -187,11 +189,82 @@ func TestCheckpointing(t *testing.T) {
 		t.Errorf("unexpected checkpoint2: %v err: %v", checkpoint2, err)
 	}
 
+	result2, err := writer.Checkpoint()
+	verifyNoError(t, err)
+
+	if result2 != result {
+		t.Errorf("invalid checkpoint after result: %v vs %v", result2, result)
+	}
+
 	verifyFull(ctx, t, om, checkpoint3, allZeroes)
 	verifyFull(ctx, t, om, checkpoint4, make([]byte, 2<<20))
 	verifyFull(ctx, t, om, result, make([]byte, 2<<20+50))
 }
 
+func TestObjectWriterRaceBetweenCheckpointAndResult(t *testing.T) {
+	rand.Seed(clock.Now().UnixNano())
+
+	ctx := testlogging.Context(t)
+	data := map[content.ID][]byte{}
+	fcm := &fakeContentManager{
+		data: data,
+	}
+
+	om, err := NewObjectManager(testlogging.Context(t), fcm, Format{
+		Splitter: "FIXED-1M",
+	}, ManagerOptions{})
+	if err != nil {
+		t.Fatalf("can't create object manager: %v", err)
+	}
+
+	allZeroes := make([]byte, 1<<20-5)
+	repeat := 100
+
+	if runtime.GOARCH == "arm" {
+		repeat = 10
+	}
+
+	for i := 0; i < repeat; i++ {
+		w := om.NewWriter(ctx, WriterOptions{
+			AsyncWrites: 1,
+		})
+
+		w.Write(allZeroes)
+		w.Write(allZeroes)
+		w.Write(allZeroes)
+
+		var eg errgroup.Group
+
+		eg.Go(func() error {
+			_, rerr := w.Result()
+
+			return rerr
+		})
+
+		eg.Go(func() error {
+			cpID, cperr := w.Checkpoint()
+			if cperr == nil && cpID != "" {
+				ids, verr := om.VerifyObject(ctx, cpID)
+				if verr != nil {
+					return errors.Wrapf(err, "Checkpoint() returned invalid object %v", cpID)
+				}
+
+				for _, id := range ids {
+					if id == "" {
+						return errors.Errorf("checkpoint returned empty id")
+					}
+				}
+			}
+
+			return nil
+		})
+
+		if err := eg.Wait(); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
 func verifyFull(ctx context.Context, t *testing.T, om *Manager, oid ID, want []byte) {
 	t.Helper()
 

repo/object/object_writer.go

@@ -83,7 +83,8 @@ type objectWriter struct {
 
 	splitter splitter.Splitter
 
-	writeMutex sync.Mutex
+	// provides mutual exclusion of all public APIs (Write, Result, Checkpoint)
+	mu sync.Mutex
 
 	asyncWritesSemaphore chan struct{} // async writes semaphore or  nil
 	asyncWritesWG        sync.WaitGroup
@@ -111,8 +112,8 @@ func (w *objectWriter) Close() error {
 }
 
 func (w *objectWriter) Write(data []byte) (n int, err error) {
-	w.writeMutex.Lock()
-	defer w.writeMutex.Unlock()
+	w.mu.Lock()
+	defer w.mu.Unlock()
 
 	dataLen := len(data)
 	w.totalLength += int64(dataLen)
@@ -240,18 +241,10 @@ func maybeCompressedContentBytes(comp compression.Compressor, output *bytes.Buff
 	return input, false, nil
 }
 
-func (w *objectWriter) drainWrites() {
-	w.writeMutex.Lock()
-
-	// wait for any in-flight asynchronous writes to finish
-	w.asyncWritesWG.Wait()
-}
-
-func (w *objectWriter) undrainWrites() {
-	w.writeMutex.Unlock()
-}
-
 func (w *objectWriter) Result() (ID, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
 	// no need to hold a lock on w.indirectIndexGrowMutex, since growing index only happens synchronously
 	// and never in parallel with calling Result()
 	if w.buffer.Len() > 0 || len(w.indirectIndex) == 0 {
@@ -260,14 +253,21 @@ func (w *objectWriter) Result() (ID, error) {
 		}
 	}
 
-	return w.Checkpoint()
+	return w.checkpointLocked()
 }
 
 // Checkpoint returns object ID which represents portion of the object that has already been written.
 // The result may be an empty object ID if nothing has been flushed yet.
 func (w *objectWriter) Checkpoint() (ID, error) {
-	w.drainWrites()
-	defer w.undrainWrites()
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	return w.checkpointLocked()
+}
+
+func (w *objectWriter) checkpointLocked() (ID, error) {
+	// wait for any in-flight asynchronous writes to finish
+	w.asyncWritesWG.Wait()
 
 	if w.contentWriteError != nil {
 		return "", w.contentWriteError

snapshot/snapshotfs/checkpoint_registry.go

@@ -51,6 +51,11 @@ func (r *checkpointRegistry) runCheckpoints(checkpointBuilder *dirManifestBuilde
 			return errors.Wrapf(err, "error checkpointing %v", n)
 		}
 
+		if de == nil {
+			// no checkpoint.
+			continue
+		}
+
 		if de.Type != snapshot.EntryTypeDirectory {
 			de.Name = ".checkpointed." + de.Name + "." + uuid.New().String()
 		}

snapshot/snapshotfs/checkpoint_registry_test.go

@@ -18,6 +18,7 @@ func TestCheckpointRegistry(t *testing.T) {
 	f1 := d.AddFile("f1", []byte{1, 2, 3}, os.FileMode(0o755))
 	f2 := d.AddFile("f2", []byte{2, 3, 4}, os.FileMode(0o755))
 	f3 := d.AddFile("f3", []byte{2, 3, 4}, os.FileMode(0o755))
+	f4 := d.AddFile("f3", []byte{2, 3, 4}, os.FileMode(0o755))
 
 	cp.addCheckpointCallback(dir1, func() (*snapshot.DirEntry, error) {
 		return &snapshot.DirEntry{
@@ -44,6 +45,10 @@ func TestCheckpointRegistry(t *testing.T) {
 		}, nil
 	})
 
+	cp.addCheckpointCallback(f4, func() (*snapshot.DirEntry, error) {
+		return nil, nil
+	})
+
 	// remove callback before it has a chance of firing
 	cp.removeCheckpointCallback(f3)
 	cp.removeCheckpointCallback(f3)

snapshot/snapshotfs/upload.go

@@ -127,6 +127,10 @@ func (u *Uploader) uploadFileInternal(ctx context.Context, parentCheckpointRegis
 			return nil, err
 		}
 
+		if checkpointID == "" {
+			return nil, nil
+		}
+
 		return newDirEntry(f, checkpointID)
 	})