fix: tests

Author: istae

fail

@@ -146,6 +146,9 @@ func (s *Service) chunksWorker(warmupTime time.Duration) {
 
 			wg.Done()
 			<-sem
+
+			s.inflight.delete(op.identityAddress, op.Chunk.Stamp().BatchID())
+
 			if reAttempt {
 				select {
 				case cc <- op:
@@ -254,12 +257,6 @@ func (s *Service) chunksWorker(warmupTime time.Duration) {
 func (s *Service) pushDeferred(ctx context.Context, logger log.Logger, op *Op) (repeat bool, err error) {
 	loggerV1 := logger.V(1).Build()
 
-	defer func() {
-		if !repeat {
-			s.inflight.delete(op.identityAddress, op.Chunk.Stamp().BatchID())
-		}
-	}()
-
 	ok, err := s.batchExist.Exists(op.Chunk.Stamp().BatchID())
 	if !ok || err != nil {
 		loggerV1.Warning(
@@ -315,9 +312,6 @@ func (s *Service) pushDirect(ctx context.Context, logger log.Logger, op *Op) (re
 		default:
 			loggerV1.Error(err, "pusher: failed to return error for direct upload")
 		}
-		if !reAttempt {
-			s.inflight.delete(op.identityAddress, op.Chunk.Stamp().BatchID())
-		}
 	}()
 
 	ok, err := s.batchExist.Exists(op.Chunk.Stamp().BatchID())

pkg/pusher/pusher.go

@@ -338,7 +338,7 @@ func TestPusherRetryShallow(t *testing.T) {
 
 	storer.chunks <- chunk
 
-	err := spinlock.Wait(spinTimeout, func() bool {
+	err := spinlock.Wait(time.Minute*5, func() bool {
 		c := int(atomic.LoadInt32(&callCount))
 		return c == retryCount
 	})

pkg/pusher/pusher_test.go

@@ -5,6 +5,7 @@
 package events
 
 import (
+	"slices"
 	"sync"
 )
 
@@ -33,7 +34,7 @@ func (b *Subscriber) Subscribe(str string) (<-chan struct{}, func()) {
 		for i, s := range b.subs[str] {
 			if s == c {
 				b.subs[str][i] = nil
-				b.subs[str] = append(b.subs[str][:i], b.subs[str][i+1:]...)
+				b.subs[str] = slices.Delete(b.subs[str], i, i+1)
 				break
 			}
 		}

pkg/storer/internal/events/subscribe.go

@@ -19,7 +19,11 @@ import (
 // swarm reference associated with this session.
 type PutterCloserWithReference interface {
 	Put(context.Context, transaction.Store, swarm.Chunk) error
-	Close(transaction.Storage, swarm.Address) error
+	Close(storage.IndexStore, swarm.Address) error
+}
+
+type PutterCloserCleanerWithReference interface {
+	PutterCloserWithReference
 	Cleanup(transaction.Storage) error
 }
 

pkg/storer/internal/internal.go

@@ -67,7 +67,7 @@ type CollectionStat struct {
 // It will create a new UUID for the collection which can be used to iterate on all the chunks
 // that are part of this collection. The root pin is only updated on successful close of this.
 // Calls to the Putter MUST be mutex locked to prevent concurrent upload data races.
-func NewCollection(st storage.IndexStore) (internal.PutterCloserWithReference, error) {
+func NewCollection(st storage.IndexStore) (internal.PutterCloserCleanerWithReference, error) {
 	newCollectionUUID := newUUID()
 	err := st.Put(&dirtyCollection{UUID: newCollectionUUID})
 	if err != nil {
@@ -121,40 +121,35 @@ func (c *collectionPutter) Put(ctx context.Context, st transaction.Store, ch swa
 	return nil
 }
 
-func (c *collectionPutter) Close(st transaction.Storage, root swarm.Address) error {
+func (c *collectionPutter) Close(st storage.IndexStore, root swarm.Address) error {
 	if root.IsZero() {
 		return errCollectionRootAddressIsZero
 	}
 
-	return st.Run(context.Background(), func(s transaction.Store) error {
-
-		collection := &pinCollectionItem{Addr: root}
-		has, err := s.IndexStore().Has(collection)
-		if err != nil {
-			return fmt.Errorf("pin store: check previous root: %w", err)
-		}
-
-		if has {
-			return ErrDuplicatePinCollection
-		}
-
-		// Save the root pin reference.
-		c.collection.Addr = root
-		err = s.IndexStore().Put(c.collection)
-		if err != nil {
-			return fmt.Errorf("pin store: failed updating collection: %w", err)
-		}
+	collection := &pinCollectionItem{Addr: root}
+	has, err := st.Has(collection)
+	if err != nil {
+		return fmt.Errorf("pin store: check previous root: %w", err)
+	}
 
-		err = s.IndexStore().Delete(&dirtyCollection{UUID: c.collection.UUID})
-		if err != nil {
-			return fmt.Errorf("pin store: failed deleting dirty collection: %w", err)
-		}
+	if has {
+		return ErrDuplicatePinCollection
+	}
 
-		c.closed = true
-		return nil
+	// Save the root pin reference.
+	c.collection.Addr = root
+	err = st.Put(c.collection)
+	if err != nil {
+		return fmt.Errorf("pin store: failed updating collection: %w", err)
+	}
 
-	})
+	err = st.Delete(&dirtyCollection{UUID: c.collection.UUID})
+	if err != nil {
+		return fmt.Errorf("pin store: failed deleting dirty collection: %w", err)
+	}
 
+	c.closed = true
+	return nil
 }
 
 func (c *collectionPutter) Cleanup(st transaction.Storage) error {

pkg/storer/internal/pinning/pinning.go

@@ -93,7 +93,9 @@ func TestPinStore(t *testing.T) {
 					}
 				}
 
-				if err := putter.Close(st, tc.root.Address()); err != nil {
+				if err := st.Run(context.Background(), func(s transaction.Store) error {
+					return putter.Close(s.IndexStore(), tc.root.Address())
+				}); err != nil {
 					t.Fatal(err)
 				}
 			})
@@ -276,7 +278,9 @@ func TestPinStore(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		err = putter.Close(st, root.Address())
+		err = st.Run(context.Background(), func(s transaction.Store) error {
+			return putter.Close(s.IndexStore(), root.Address())
+		})
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -311,12 +315,16 @@ func TestPinStore(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		err = putter.Close(st, root.Address())
+		err = st.Run(context.Background(), func(s transaction.Store) error {
+			return putter.Close(s.IndexStore(), root.Address())
+		})
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		err = putter.Close(st, root.Address())
+		err = st.Run(context.Background(), func(s transaction.Store) error {
+			return putter.Close(s.IndexStore(), root.Address())
+		})
 		if err == nil || !errors.Is(err, pinstore.ErrDuplicatePinCollection) {
 			t.Fatalf("unexpected error during CLose, want: %v, got: %v", pinstore.ErrDuplicatePinCollection, err)
 		}
@@ -344,7 +352,9 @@ func TestPinStore(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		err = putter.Close(st, swarm.ZeroAddress)
+		err = st.Run(context.Background(), func(s transaction.Store) error {
+			return putter.Close(s.IndexStore(), swarm.ZeroAddress)
+		})
 		if !errors.Is(err, pinstore.ErrCollectionRootAddressIsZero) {
 			t.Fatalf("unexpected error on close, want: %v, got: %v", pinstore.ErrCollectionRootAddressIsZero, err)
 		}
@@ -375,7 +385,7 @@ func TestCleanup(t *testing.T) {
 		chunks := chunktest.GenerateTestRandomChunks(5)
 
 		var (
-			putter internal.PutterCloserWithReference
+			putter internal.PutterCloserCleanerWithReference
 			err    error
 		)
 		err = st.Run(context.Background(), func(s transaction.Store) error {

pkg/storer/internal/pinning/pinning_test.go

@@ -454,62 +454,46 @@ func (u *uploadPutter) Put(ctx context.Context, st transaction.Store, chunk swar
 // with a swarm reference. This can be useful while keeping track of uploads through
 // the tags. It will update the tag. This will be filled with the Split and Seen count
 // by the Putter.
-func (u *uploadPutter) Close(st transaction.Storage, addr swarm.Address) error {
+func (u *uploadPutter) Close(s storage.IndexStore, addr swarm.Address) error {
 	if u.closed {
 		return nil
 	}
-	defer func() {
-		u.closed = true
-	}()
 
-	if err := u.Cleanup(st); err != nil {
-		return fmt.Errorf("cleanup failed on close, root_ref %s: %w", addr, err)
-	}
-
-	return st.Run(context.Background(), func(s transaction.Store) error {
+	u.closed = true
 
-		ti := &TagItem{TagID: u.tagID}
-		err := s.IndexStore().Get(ti)
-		if err != nil {
-			if errors.Is(err, storage.ErrNotFound) {
-				return nil
-			}
-			return fmt.Errorf("failed reading tag while closing: %w", err)
+	ti := &TagItem{TagID: u.tagID}
+	err := s.Get(ti)
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			return s.Delete(&dirtyTagItem{TagID: u.tagID})
 		}
+		return fmt.Errorf("failed reading tag while closing: %w", err)
+	}
 
-		ti.Split += u.split
-		ti.Seen += u.seen
+	ti.Split += u.split
+	ti.Seen += u.seen
 
-		if !addr.IsZero() {
-			ti.Address = addr.Clone()
-		}
+	if !addr.IsZero() {
+		ti.Address = addr.Clone()
+	}
 
-		return s.IndexStore().Put(ti)
-	})
+	return errors.Join(
+		s.Put(ti),
+		s.Delete(&dirtyTagItem{TagID: u.tagID}),
+	)
 }
 
-func (u *uploadPutter) Cleanup(st transaction.Storage) error {
-	if u.closed {
-		return errPutterAlreadyClosed
-	}
+func Cleanup(st transaction.Storage, tag uint64) error {
 
 	itemsToDelete := make([]*pushItem, 0)
 
-	di := &dirtyTagItem{TagID: u.tagID}
-	err := st.IndexStore().Get(di)
-	if err != nil {
-		return fmt.Errorf("failed reading dirty tag while cleaning up: %w", err)
-	}
-
-	err = st.IndexStore().Iterate(
+	err := st.IndexStore().Iterate(
 		storage.Query{
-			Factory:       func() storage.Item { return &pushItem{} },
-			PrefixAtStart: true,
-			Prefix:        fmt.Sprintf("%d", di.Started),
+			Factory: func() storage.Item { return &pushItem{} },
 		},
 		func(res storage.Result) (bool, error) {
 			pi := res.Entry.(*pushItem)
-			if pi.TagID == u.tagID {
+			if pi.TagID == tag {
 				itemsToDelete = append(itemsToDelete, pi)
 			}
 			return false, nil
@@ -540,7 +524,7 @@ func (u *uploadPutter) Cleanup(st transaction.Storage) error {
 	return errors.Join(
 		eg.Wait(),
 		st.Run(context.Background(), func(s transaction.Store) error {
-			return s.IndexStore().Delete(di)
+			return s.IndexStore().Delete(&dirtyTagItem{TagID: tag})
 		}),
 	)
 }
@@ -564,33 +548,33 @@ func CleanupDirty(st transaction.Storage) error {
 	}
 
 	for _, di := range dirtyTags {
-		err = errors.Join(err, (&uploadPutter{tagID: di.TagID}).Cleanup(st))
+		err = errors.Join(err, Cleanup(st, di.TagID))
 	}
 
 	return err
 }
 
 // Report is the implementation of the PushReporter interface.
 // Must be mutex locked.
-func Report(st transaction.Store, tag uint64, update *TagUpdate) error {
+func Report(st storage.IndexStore, tag uint64, update *TagUpdate) (bool, error) {
 
 	ti := &TagItem{TagID: tag}
-	err := st.IndexStore().Get(ti)
+	err := st.Get(ti)
 	if err != nil {
 		if !errors.Is(err, storage.ErrNotFound) {
-			return fmt.Errorf("failed getting tag: %w", err)
+			return false, fmt.Errorf("failed getting tag: %w", err)
 		}
 
 		// tag is missing, no need update it
-		return nil
+		return false, nil
 	}
 
 	// update the tag
 	ti.Sent += update.Sent
 	ti.Stored += update.Stored
 	ti.Synced += update.Synced
 
-	return st.IndexStore().Put(ti)
+	return ti.Synced >= ti.Split, st.Put(ti)
 }
 
 var (
@@ -683,9 +667,11 @@ func ListAllTags(st storage.Reader) ([]TagItem, error) {
 	return tags, nil
 }
 
-func IteratePending(ctx context.Context, s transaction.ReadOnlyStore, consumerFn func(chunk swarm.Chunk) (bool, error)) error {
+func IteratePending(ctx context.Context, s transaction.ReadOnlyStore, start int64, consumerFn func(chunk swarm.Chunk, ts int64) (bool, error)) error {
 	return s.IndexStore().Iterate(storage.Query{
-		Factory: func() storage.Item { return &pushItem{} },
+		Factory:       func() storage.Item { return &pushItem{} },
+		PrefixAtStart: true,
+		Prefix:        fmt.Sprintf("%d", start),
 	}, func(r storage.Result) (bool, error) {
 		pi := r.Entry.(*pushItem)
 		has, err := s.IndexStore().Has(&dirtyTagItem{TagID: pi.TagID})
@@ -709,7 +695,7 @@ func IteratePending(ctx context.Context, s transaction.ReadOnlyStore, consumerFn
 			WithStamp(stamp).
 			WithTagID(pi.TagID)
 
-		return consumerFn(chunk)
+		return consumerFn(chunk, pi.Timestamp)
 	})
 }