Close all dechunkers at Store shutdown

command/chunking/dechunker.go

@@ -117,5 +117,19 @@ func (d *DechunkerManager) Get(id string) (*Dechunker, error) {
 func (d *DechunkerManager) Delete(id string) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
-	delete(d.m, id)
+	dc, ok := d.m[id]
+	if ok {
+		dc.Close()
+		delete(d.m, id)
+	}
+}
+
+// Closes closes the DechunkerManager and all Dechunkers it manages.
+func (d *DechunkerManager) Close() {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	for _, dc := range d.m {
+		dc.Close()
+		delete(d.m, dc.streamID)
+	}
 }

command/chunking/dechunker_test.go

@@ -315,6 +315,7 @@ func Test_CreateDechunkerManager(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to create DechunkerManager: %v", err)
 	}
+	defer manager.Close()
 	if manager == nil {
 		t.Fatalf("expected DechunkerManager instance, got nil")
 	}
@@ -342,6 +343,7 @@ func Test_GetDechunker(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to create DechunkerManager: %v", err)
 	}
+	defer manager.Close()
 
 	dechunker1, err := manager.Get("test1")
 	if err != nil {
@@ -370,6 +372,7 @@ func Test_DeleteDechunker(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to create DechunkerManager: %v", err)
 	}
+	defer manager.Close()
 
 	dechunker1, err := manager.Get("test1")
 	if err != nil {

store/store.go

@@ -587,6 +587,8 @@ func (s *Store) Close(wait bool) (retErr error) {
 	}
 	s.logger.Printf("closing store with node ID %s, listening on %s", s.raftID, s.ln.Addr().String())
 
+	s.dechunkManager.Close()
+
 	close(s.appliedIdxUpdateDone)
 	close(s.observerClose)
 	<-s.observerDone