From 2c01fd8bf846c178d11130ad58d345b336615963 Mon Sep 17 00:00:00 2001 From: Gilles Chehade Date: Sun, 24 May 2026 01:10:53 +0200 Subject: [PATCH] packer: don't panic on Put after Wait When Wait closed packerChan, any in-flight Put would panic with "send on closed channel". Snapshot.Backup has error paths that call PackerManager.Wait while workers are still calling Put (e.g. recordError on a Ctrl+C-interrupted backup), so the panic surfaced in real use. Switch from closing packerChan to a separate closing channel: Put selects on either the send or closing and returns ErrShutdown on shutdown. Run workers also observe closing and flush their in-flight packfile before returning. Wait is now idempotent. Same fix applied to the platar packer for consistency. Adds two regression tests: - TestSeqPackerPutDuringWait races many concurrent Put callers against Wait. Hits "send on closed channel" reliably without the fix. - TestSeqPackerPutAfterWait pins the post-shutdown contract: Put returns an error instead of panicking. Fixes PlakarKorp/plakar#2106 Co-Authored-By: Claude Sonnet 4.6 --- repository/packer/packer_race_test.go | 104 ++++++++++++++++++++++++ repository/packer/packer_seq.go | 109 ++++++++++++++------------ repository/packer/platar.go | 18 ++++- 3 files changed, 176 insertions(+), 55 deletions(-) create mode 100644 repository/packer/packer_race_test.go diff --git a/repository/packer/packer_race_test.go b/repository/packer/packer_race_test.go new file mode 100644 index 000000000..d093c9b83 --- /dev/null +++ b/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") + } +} diff --git a/repository/packer/packer_seq.go b/repository/packer/packer_seq.go index 1a8f94852..aeec2f21b 100644 --- a/repository/packer/packer_seq.go +++ b/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,12 +175,10 @@ 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. @@ -170,40 +186,17 @@ func (mgr *seqPackerManager) Run() error { 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 } } diff --git a/repository/packer/platar.go b/repository/packer/platar.go index aa6844cf7..0ac1dd67e 100644 --- a/repository/packer/platar.go +++ b/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) {