packer: don't panic on Put after Wait

repository/packer/packer_race_test.go

@@ -0,0 +1,104 @@
+package packer_test
+
+import (
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/PlakarKorp/kloset/connectors/storage"
+	"github.com/PlakarKorp/kloset/kcontext"
+	"github.com/PlakarKorp/kloset/objects"
+	"github.com/PlakarKorp/kloset/packfile"
+	"github.com/PlakarKorp/kloset/repository/packer"
+	"github.com/PlakarKorp/kloset/resources"
+)
+
+// TestSeqPackerPutDuringWait is the regression test for
+// https://github.com/PlakarKorp/plakar/issues/2106
+//
+// Before the fix, Put sent to packerChan unconditionally; Wait closed those
+// channels. Any goroutine that called Put after Wait had begun would
+// "panic: send on closed channel". After the fix, Put must return an error
+// instead of panicking when the manager has been shut down.
+//
+// The test launches many concurrent Put callers and races them against a
+// Wait. Done under -race -count=N to surface the panic reliably.
+func TestSeqPackerPutDuringWait(t *testing.T) {
+	const producers = 32
+	const putsPerProducer = 200
+
+	ctx := kcontext.NewKContext()
+	ctx.MaxConcurrency = runtime.NumCPU()
+	storageConf := storage.NewConfiguration()
+
+	flusher := func(pf packfile.Packfile) error { return nil }
+	mgr := packer.NewSeqPackerManager(ctx, storageConf, identityEncoder, newTestPackfileFactory(), sha256Factory, flusher)
+
+	runErr := make(chan error, 1)
+	go func() { runErr <- mgr.Run() }()
+
+	// Producers continuously put blobs. They must never panic, even if they
+	// happen to call Put after Wait has closed the channels.
+	var producersDone sync.WaitGroup
+	var panics atomic.Int32
+	for p := range producers {
+		producersDone.Add(1)
+		go func(seed int) {
+			defer producersDone.Done()
+			defer func() {
+				if r := recover(); r != nil {
+					panics.Add(1)
+					t.Errorf("producer %d panicked: %v", seed, r)
+				}
+			}()
+			for i := range putsPerProducer {
+				mac := objects.MAC{byte(seed), byte(i), byte(i >> 8)}
+				// Ignore the error: after shutdown Put will return one, which
+				// is the contract under test.
+				_ = mgr.Put(-1, resources.RT_CHUNK, mac, []byte("payload"))
+			}
+		}(p)
+	}
+
+	// Give the producers a moment to ramp up so that a meaningful subset of
+	// them are mid-flight when Wait fires, then shut down.
+	runtime.Gosched()
+	mgr.Wait()
+	producersDone.Wait()
+
+	if err := <-runErr; err != nil {
+		t.Fatalf("Run returned error: %v", err)
+	}
+	if got := panics.Load(); got != 0 {
+		t.Fatalf("producers panicked %d times; expected 0", got)
+	}
+}
+
+// TestSeqPackerPutAfterWait pins the explicit post-shutdown contract: once
+// Wait has returned, any subsequent Put must return an error rather than
+// panic.
+func TestSeqPackerPutAfterWait(t *testing.T) {
+	ctx := kcontext.NewKContext()
+	ctx.MaxConcurrency = 2
+	storageConf := storage.NewConfiguration()
+	flusher := func(pf packfile.Packfile) error { return nil }
+	mgr := packer.NewSeqPackerManager(ctx, storageConf, identityEncoder, newTestPackfileFactory(), sha256Factory, flusher)
+
+	runErr := make(chan error, 1)
+	go func() { runErr <- mgr.Run() }()
+	mgr.Wait()
+	if err := <-runErr; err != nil {
+		t.Fatalf("Run returned error: %v", err)
+	}
+
+	defer func() {
+		if r := recover(); r != nil {
+			t.Fatalf("Put after Wait panicked: %v", r)
+		}
+	}()
+	mac := objects.MAC{99}
+	if err := mgr.Put(-1, resources.RT_CHUNK, mac, []byte("x")); err == nil {
+		t.Fatalf("expected Put after Wait to return an error, got nil")
+	}
+}

repository/packer/packer_seq.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
+	"errors"
 	"fmt"
 	"hash"
 	"io"
@@ -22,11 +23,20 @@ import (
 	"github.com/PlakarKorp/kloset/versioning"
 )
 
+// ErrShutdown is returned by Put when Wait has been called. See
+// PlakarKorp/plakar#2106.
+var ErrShutdown = errors.New("packer: manager has been shut down")
+
 type seqPackerManager struct {
 	inflightMACs   map[resources.Type]*sync.Map
 	packerChan     []chan PackerMsg
 	packerChanDone chan struct{}
 
+	// closing is closed by Wait to signal shutdown to both Run workers and
+	// in-flight Put callers (which select on it instead of sending blindly).
+	closing   chan struct{}
+	closeOnce sync.Once
+
 	packfileFactory packfile.PackfileCtor
 	storageConf     *storage.Configuration
 	encodingFunc    func(io.Reader) (io.Reader, error)
@@ -51,6 +61,7 @@ func NewSeqPackerManager(ctx *kcontext.KContext, storageConfiguration *storage.C
 		inflightMACs:    inflightsMACs,
 		packerChan:      make([]chan PackerMsg, nChan),
 		packerChanDone:  make(chan struct{}),
+		closing:         make(chan struct{}),
 		packfileFactory: packfileFactory,
 		storageConf:     storageConfiguration,
 		encodingFunc:    encodingFunc,
@@ -112,20 +123,27 @@ func (mgr *seqPackerManager) Run() error {
 		workerGroup.Go(func() error {
 			var pfile packfile.Packfile
 
+			// flushTail finalizes the in-flight packfile, if any, before
+			// the worker returns.
+			flushTail := func() error {
+				if pfile != nil && pfile.Size() > 0 {
+					if err := mgr.AddPadding(pfile, int(mgr.storageConf.Chunking.MinSize)); err != nil {
+						return err
+					}
+					packerResultChan <- pfile
+				}
+				return nil
+			}
+
 			for {
 				select {
 				case <-workerCtx.Done():
 					return workerCtx.Err()
+				case <-mgr.closing:
+					return flushTail()
 				case pm, ok := <-mgr.packerChan[i]:
 					if !ok {
-						if pfile != nil && pfile.Size() > 0 {
-							if err := mgr.AddPadding(pfile, int(mgr.storageConf.Chunking.MinSize)); err != nil {
-								return err
-							}
-
-							packerResultChan <- pfile
-						}
-						return nil
+						return flushTail()
 					}
 
 					if pfile == nil {
@@ -157,53 +175,28 @@ func (mgr *seqPackerManager) Run() error {
 		})
 	}
 
-	inError := false
 	// Wait for workers to finish.
 	if err := workerGroup.Wait(); err != nil {
 		mgr.appCtx.GetLogger().Error("Worker group error: %s", err)
 		mgr.appCtx.Cancel(err)
-		inError = true
 	}
 
 	// Close the result channel and wait for the flusher to finish.
 	close(packerResultChan)
 	if err := flusherGroup.Wait(); err != nil {
 		mgr.appCtx.GetLogger().Error("Flusher group error: %s", err)
 		mgr.appCtx.Cancel(err)
-		inError = true
 	}
 
-	if inError {
-		// If we are in error we have to wait for the producers to catch up for
-		// the Cancellation meaning we still have to drain the channels.
-		drainingGroup := sync.WaitGroup{}
-		for i := range mgr.nChan {
-			drainingGroup.Go(func() {
-				for {
-					select {
-					case _, ok := <-mgr.packerChan[i]:
-						if !ok {
-							return
-						}
-					}
-				}
-			})
-		}
-
-		drainingGroup.Wait()
-	}
-
-	// Signal completion.
 	mgr.packerChanDone <- struct{}{}
 	close(mgr.packerChanDone)
 	return nil
 }
 
 func (mgr *seqPackerManager) Wait() {
-	for i := range mgr.nChan {
-		close(mgr.packerChan[i])
-	}
-
+	// closeOnce makes Wait safe to call more than once; some error paths in
+	// snapshot.Backup do call it twice.
+	mgr.closeOnce.Do(func() { close(mgr.closing) })
 	<-mgr.packerChanDone
 }
 
@@ -217,26 +210,38 @@ func (mgr *seqPackerManager) InsertIfNotPresent(Type resources.Type, mac objects
 }
 
 func (mgr *seqPackerManager) Put(hint int, Type resources.Type, mac objects.MAC, data []byte) error {
-	if encodedReader, err := mgr.encodingFunc(bytes.NewReader(data)); err != nil {
+	// Skip encoding work if shutdown was already signalled.
+	select {
+	case <-mgr.closing:
+		return ErrShutdown
+	default:
+	}
+
+	encodedReader, err := mgr.encodingFunc(bytes.NewReader(data))
+	if err != nil {
 		return err
-	} else {
-		encoded, err := io.ReadAll(encodedReader)
-		if err != nil {
-			return err
-		}
-
-		if Type != resources.RT_OBJECT && Type != resources.RT_CHUNK {
-			mgr.packerChan[mgr.nChan-1] <- PackerMsg{Type: Type, Version: versioning.GetCurrentVersion(Type), Timestamp: time.Now(), MAC: mac, Data: encoded}
-		} else {
-			if hint == -1 {
-				mgr.packerChan[randv2.IntN(mgr.nChan-1)] <- PackerMsg{Type: Type, Version: versioning.GetCurrentVersion(Type), Timestamp: time.Now(), MAC: mac, Data: encoded}
-			} else {
-				mgr.packerChan[hint] <- PackerMsg{Type: Type, Version: versioning.GetCurrentVersion(Type), Timestamp: time.Now(), MAC: mac, Data: encoded}
-			}
+	}
+	encoded, err := io.ReadAll(encodedReader)
+	if err != nil {
+		return err
+	}
 
-		}
+	var i int
+	switch {
+	case Type != resources.RT_OBJECT && Type != resources.RT_CHUNK:
+		i = mgr.nChan - 1
+	case hint == -1:
+		i = randv2.IntN(mgr.nChan - 1)
+	default:
+		i = hint
+	}
 
+	msg := PackerMsg{Type: Type, Version: versioning.GetCurrentVersion(Type), Timestamp: time.Now(), MAC: mac, Data: encoded}
+	select {
+	case mgr.packerChan[i] <- msg:
 		return nil
+	case <-mgr.closing:
+		return ErrShutdown
 	}
 }
 

repository/packer/platar.go

@@ -6,6 +6,7 @@ import (
 	"hash"
 	"io"
 	"runtime"
+	"sync"
 	"time"
 
 	"golang.org/x/sync/errgroup"
@@ -24,6 +25,9 @@ type platarPackerManager struct {
 	packerChan     chan interface{}
 	packerChanDone chan struct{}
 
+	closing   chan struct{}
+	closeOnce sync.Once
+
 	storageConf  *storage.Configuration
 	encodingFunc func(io.Reader) (io.Reader, error)
 	hashFactory  func() hash.Hash
@@ -44,6 +48,7 @@ func NewPlatarPackerManager(ctx *kcontext.KContext, storageConfiguration *storag
 		packingCache:   cache,
 		packerChan:     make(chan interface{}, runtime.NumCPU()*2+1),
 		packerChanDone: make(chan struct{}),
+		closing:        make(chan struct{}),
 		storageConf:    storageConfiguration,
 		encodingFunc:   encodingFunc,
 		hashFactory:    hashFactory,
@@ -64,6 +69,8 @@ func (mgr *platarPackerManager) Run() error {
 				select {
 				case <-workerCtx.Done():
 					return workerCtx.Err()
+				case <-mgr.closing:
+					return nil
 				case msg, ok := <-mgr.packerChan:
 					if !ok {
 						return nil
@@ -103,7 +110,7 @@ func (mgr *platarPackerManager) Run() error {
 }
 
 func (mgr *platarPackerManager) Wait() {
-	close(mgr.packerChan)
+	mgr.closeOnce.Do(func() { close(mgr.closing) })
 	<-mgr.packerChanDone
 }
 
@@ -126,8 +133,13 @@ func (mgr *platarPackerManager) InsertIfNotPresent(Type resources.Type, mac obje
 }
 
 func (mgr *platarPackerManager) Put(_ int, Type resources.Type, mac objects.MAC, data []byte) error {
-	mgr.packerChan <- &PackerMsg{Type: Type, Version: versioning.GetCurrentVersion(Type), Timestamp: time.Now(), MAC: mac, Data: data}
-	return nil
+	msg := &PackerMsg{Type: Type, Version: versioning.GetCurrentVersion(Type), Timestamp: time.Now(), MAC: mac, Data: data}
+	select {
+	case mgr.packerChan <- msg:
+		return nil
+	case <-mgr.closing:
+		return ErrShutdown
+	}
 }
 
 func (mgr *platarPackerManager) Exists(Type resources.Type, mac objects.MAC) (bool, error) {