fix: make stamps removal sync and data race in push chunk tests

pkg/api/gsoc_test.go

@@ -35,7 +35,7 @@ func TestGsocWebsocketSingleHandler(t *testing.T) {
 		payload          = []byte("hello there!")
 	)
 
-	err := cl.SetReadDeadline(time.Now().Add(2 * time.Second))
+	err := cl.SetReadDeadline(time.Now().Add(longTimeout))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -70,7 +70,7 @@ func TestGsocWebsocketMultiHandler(t *testing.T) {
 	}
 	testutil.CleanupCloser(t, cl2)
 
-	err = cl.SetReadDeadline(time.Now().Add(2 * time.Second))
+	err = cl.SetReadDeadline(time.Now().Add(longTimeout))
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/p2p/streamtest/streamtest.go

@@ -401,7 +401,11 @@ func (r *record) close() {
 }
 
 func (r *record) bytes() []byte {
-	return r.b
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	cp := make([]byte, len(r.b))
+	copy(cp, r.b)
+	return cp
 }
 
 func (r *record) bytesSize() int {

pkg/postage/service.go

@@ -290,37 +290,39 @@ func (ps *service) HandleStampExpiry(ctx context.Context, id []byte) error {
 	return nil
 }
 
-// removeStampItems
+// removeStampItems removes all stamp items belonging to the given batch.
 func (ps *service) removeStampItems(ctx context.Context, batchID []byte) error {
 	ps.logger.Debug("removing expired stamp items", "batchID", hex.EncodeToString(batchID))
 
-	deleteItemC := make(chan *StampItem)
-	go func() {
-		for item := range deleteItemC {
-			_ = ps.store.Delete(item)
-		}
-	}()
-
-	count := 0
-
-	defer func() {
-		close(deleteItemC)
-		ps.logger.Debug("removed expired stamps", "batchID", hex.EncodeToString(batchID), "count", count)
-	}()
+	var toDelete []*StampItem
 
-	return ps.store.Iterate(
+	err := ps.store.Iterate(
 		storage.Query{
 			Factory: func() storage.Item { return new(StampItem) },
 			Prefix:  string(batchID),
 		}, func(result storage.Result) (bool, error) {
-			select {
-			case deleteItemC <- result.Entry.(*StampItem):
-			case <-ctx.Done():
-				return false, ctx.Err()
+			if err := ctx.Err(); err != nil {
+				return false, err
 			}
-			count++
+			toDelete = append(toDelete, result.Entry.(*StampItem))
 			return false, nil
 		})
+	if err != nil {
+		return err
+	}
+
+	var firstErr error
+	for _, item := range toDelete {
+		if err := ps.store.Delete(item); err != nil && firstErr == nil {
+			firstErr = fmt.Errorf("remove expired stamp items batch %s: %w", hex.EncodeToString(batchID), err)
+		}
+	}
+	if firstErr != nil {
+		return firstErr
+	}
+
+	ps.logger.Debug("removed expired stamps", "batchID", hex.EncodeToString(batchID), "count", len(toDelete))
+	return nil
 }
 
 // SetExpired removes all expired batches from the stamp issuers.

pkg/postage/service_test.go

@@ -296,7 +296,7 @@ func TestSetExpired(t *testing.T) {
 
 	err = store.Get(itemNotExists)
 	if err == nil {
-		t.Fatal(err)
+		t.Fatal("expected error getting expired stamp item, got nil")
 	}
 
 	testutil.CleanupCloser(t, ps)

pkg/storer/internal/events/subscribe_test.go

@@ -45,8 +45,9 @@ func TestSubscriber(t *testing.T) {
 
 		bin1, unsub1 := s.Subscribe("1")
 		go s.Trigger("1")
-		go s.Trigger("1")
 		<-bin1
+
+		go s.Trigger("1")
 		<-bin1
 
 		unsub1()