storage: fix pyramid chunker and hasherstore possible deadlocks (ethersphere#1679)

* storage: prevent deadlock on hasherstore

* storage: pyramid chunker quit only once

Author: acud

storage/hasherstore.go

@@ -152,7 +152,10 @@ func (h *hasherStore) startWait(ctx context.Context) {
 		select {
 		// if context is done earlier, just return with the error
 		case <-ctx.Done():
-			h.waitC <- ctx.Err()
+			select {
+			case h.waitC <- ctx.Err():
+			case <-h.quitC:
+			}
 			return
 		// doneC is closed if all chunks have been submitted, from then we just wait until all of them are also stored
 		case <-doneC:
@@ -161,7 +164,10 @@ func (h *hasherStore) startWait(ctx context.Context) {
 		// a chunk has been stored, if err is nil, then successfully, so increase the stored chunk counter
 		case err := <-h.errC:
 			if err != nil {
-				h.waitC <- err
+				select {
+				case h.waitC <- err:
+				case <-h.quitC:
+				}
 				return
 			}
 			nrStoredChunks++

storage/pyramid.go

@@ -149,6 +149,7 @@ type PyramidChunker struct {
 	wg          *sync.WaitGroup
 	errC        chan error
 	quitC       chan bool
+	quitOnce    sync.Once
 	rootAddress []byte
 	chunkLevel  [][]*TreeEntry
 }
@@ -218,7 +219,7 @@ func (pc *PyramidChunker) Split(ctx context.Context) (k Address, wait func(conte
 		close(pc.errC)
 	}()
 
-	defer close(pc.quitC)
+	defer pc.quit()
 	defer pc.putter.Close()
 
 	select {
@@ -250,7 +251,7 @@ func (pc *PyramidChunker) Append(ctx context.Context) (k Address, wait func(cont
 		close(pc.errC)
 	}()
 
-	defer close(pc.quitC)
+	defer pc.quit()
 	defer pc.putter.Close()
 
 	select {
@@ -469,7 +470,7 @@ func (pc *PyramidChunker) prepareChunks(ctx context.Context, isAppend bool) {
 					break
 				}
 			} else {
-				close(pc.quitC)
+				pc.quit()
 				break
 			}
 		}
@@ -704,3 +705,9 @@ func (pc *PyramidChunker) cleanChunkLevels() {
 		}
 	}
 }
+
+func (pc *PyramidChunker) quit() {
+	pc.quitOnce.Do(func() {
+		close(pc.quitC)
+	})
+}