Do not call t.Fatal() from non-main goroutines (#291)

Calling t.Fatal() only stops the current goroutine, but not the rest.
The result is different depending on a test, but in the worst case
scenario (like an error in the server part of a test) the client may
hang waiting indefinitely for the response, and the test ends with
a panic from the test timeout, without showing the real reason for
the issue.

Author: matshch

failover_test.go

@@ -1,13 +1,14 @@
 package desync
 
 import (
+	"context"
 	"crypto/rand"
-	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 )
 
 func TestFailoverMissingChunk(t *testing.T) {
@@ -77,48 +78,46 @@ func TestFailoverMutliple(t *testing.T) {
 	g := NewFailoverGroup(storeA, storeB)
 
 	var (
-		wg       sync.WaitGroup
-		done     = make(chan struct{})
-		timeout  = time.After(time.Second)
-		failOver = time.Tick(10 * time.Millisecond)
+		ctx, cancel = context.WithTimeout(t.Context(), time.Second)
+		eg, gCtx    = errgroup.WithContext(ctx)
+		failOver    = time.Tick(10 * time.Millisecond)
 	)
+	defer cancel()
 
 	// Run several goroutines querying the group in a tight loop
 	for i := 0; i < 16; i++ {
-		wg.Add(1)
-		go func() {
+		eg.Go(func() error {
 			var id ChunkID
 			for {
 				time.Sleep(time.Millisecond)
 				select {
-				case <-done:
-					wg.Done()
-					return
+				case <-gCtx.Done():
+					return nil
 				default:
 					rand.Read(id[:])
 					if _, err := g.GetChunk(id); err != nil {
-						t.Fatal(err)
+						return err
 					}
 				}
 			}
-		}()
+		})
 	}
 
 	// Make the stores fail over every 10 ms
-	go func() {
-		wg.Add(1)
+	eg.Go(func() error {
 		for {
 			select {
-			case <-timeout: // done running
-				close(done)
-				wg.Done()
-				return
+			case <-gCtx.Done(): // done running
+				return nil
 			case <-failOver: // switch over to the other store
 				newX := (x + 1) % 2
 				atomic.StoreInt64(&x, newX)
 			}
 		}
-	}()
+	})
 
-	wg.Wait()
+	err := eg.Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
 }

mount-index_linux_test.go

@@ -55,13 +55,18 @@ func TestMountIndex(t *testing.T) {
 	}()
 
 	// Start the Fuse mount
+	c := make(chan error, 1)
 	go func() {
 		ifs := NewIndexMountFS(index, "blob1", s)
-		MountIndex(ctx, index, ifs, mnt, s, 10)
+		c <- MountIndex(ctx, index, ifs, mnt, s, 10)
 		wg.Done()
 	}()
 
-	time.Sleep(time.Second)
+	select {
+	case err = <-c:
+		t.Fatal(err)
+	case <-time.After(time.Second):
+	}
 
 	// Calculate the hash of the file in the mount point
 	b, err = ioutil.ReadFile(filepath.Join(mnt, "blob1"))

protocol_test.go

@@ -2,8 +2,12 @@ package desync
 
 import (
 	"bytes"
+	"context"
+	"errors"
 	"io"
 	"testing"
+
+	"golang.org/x/sync/errgroup"
 )
 
 func TestProtocol(t *testing.T) {
@@ -19,51 +23,74 @@ func TestProtocol(t *testing.T) {
 	compressed, _ := Compressor{}.toStorage(uncompressed)
 	cID := inChunk.ID()
 
+	ctx, cancel := context.WithCancel(t.Context())
+	g, gCtx := errgroup.WithContext(ctx)
+	defer cancel()
+
 	// Server
-	go func() {
+	g.Go(func() error {
 		flags, err := client.Initialize(CaProtocolReadableStore)
 		if err != nil {
-			t.Fatal(err)
+			return err
 		}
 		if flags&CaProtocolPullChunks == 0 {
-			t.Fatalf("client not asking for chunks")
+			return errors.New("client not asking for chunks")
 		}
 		for {
 			m, err := client.ReadMessage()
 			if err != nil {
-				t.Fatal(err)
+				if errors.Is(ctx.Err(), context.Canceled) {
+					return nil
+				}
+				return err
 			}
 			switch m.Type {
 			case CaProtocolRequest:
 				id, err := ChunkIDFromSlice(m.Body[8:40])
 				if err != nil {
-					t.Fatal(err)
+					return err
 				}
 				if err := client.SendProtocolChunk(id, 0, compressed); err != nil {
-					t.Fatal(err)
+					return err
 				}
 			default:
 
-				t.Fatal("unexpected message")
+				return errors.New("unexpected message")
 			}
 		}
-	}()
+	})
 
 	// Client
-	flags, err := server.Initialize(CaProtocolPullChunks)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if flags&CaProtocolReadableStore == 0 {
-		t.Fatalf("server not offering chunks")
-	}
+	g.Go(func() error {
+		defer cancel()
+		flags, err := server.Initialize(CaProtocolPullChunks)
+		if err != nil {
+			return err
+		}
+		if flags&CaProtocolReadableStore == 0 {
+			return errors.New("server not offering chunks")
+		}
+
+		chunk, err := server.RequestChunk(cID)
+		if err != nil {
+			return err
+		}
+		b, _ := chunk.Data()
+		if !bytes.Equal(b, uncompressed) {
+			return errors.New("chunk data doesn't match expected")
+		}
+		return nil
+	})
 
-	chunk, err := server.RequestChunk(cID)
+	<-gCtx.Done()
+	// unblock client/server in case of an error
+	r1.Close()
+	r2.Close()
+	w1.Close()
+	w2.Close()
+
+	err := g.Wait()
 	if err != nil {
 		t.Fatal(err)
 	}
-	b, _ := chunk.Data()
-	if !bytes.Equal(b, uncompressed) {
-		t.Fatal("chunk data doesn't match expected")
-	}
 }