optimized tests

TODO.md

@@ -10,16 +10,14 @@
 
 - [ ] test suite
   - [ ] add test more precise test for engine based on test data from test/data/745.squashfs
-    - [ ] this means add more squashfs samples as new cycles are produced
-  - [x] store test data in squashfs
 
 - [ ] deploy
   - [ ] prepare nginx config (I'll do this)
 
 - [x] found bugs
   - [x] tz1gXWW1q8NcXtVy2oVVcc2s4XKNzv9CryWd - 746 - minimum not found
   - [x] tz1bZ8vsMAXmaWEV7FRnyhcuUs2fYMaQ6Hkk - 746 - minimum not found
-  - [ ] NOTE: waiting for confirmation from NL engineers (should be tomorrow)
+  - [ ] NOTE: waiting for confirmation from NL engineers
 
 - [x] rolling/archive mode
   - [x] configurable through configuration
@@ -29,6 +27,4 @@
 - [x] add consensus round delay - e.g. if 2 api returns for cycle 747 results of cycle 745 
   - [x] we should fetch it from network parameters when starting because it can be different on testnets
 
-- [ ] configurable subset of delegates to track instead of all so people can run their mirrors with reduced overhead
-- [x] webhook notifications for discord - notifies about errors fetching delegates
-  - NOTE: FetchCycleDelegationStates - after `e.logger.Error("failed to fetch delegate delegation state", "cycle", cycle, "delegate", item.String(), "error", err.Error())`
+- [ ] configurable subset of delegates to track instead of all so people can run their mirrors with reduced overhead

core/collector_test.go

@@ -2,7 +2,6 @@ package core
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 	"runtime/debug"
 	"sync"
@@ -19,7 +18,7 @@ var (
 )
 
 func getTransport() *test.TestTransport {
-	transport, err := test.NewTestTransport(http.DefaultTransport, "../test/data/745", "../test/data/745.tar.xz")
+	transport, err := test.NewTestTransport(http.DefaultTransport, "../test/data/745", "../test/data/745.gob.lz4")
 	if err != nil {
 		panic(err)
 	}
@@ -50,31 +49,19 @@ func TestGetDelegationStateNoStaking(t *testing.T) {
 	delegates, err := collector.GetActiveDelegatesFromCycle(defaultCtx, cycle)
 	assert.Nil(err)
 
-	channels := make(chan error, len(delegates))
-	var wg sync.WaitGroup
-	wg.Add(len(delegates))
-	for _, addr := range delegates {
-		go func(addr tezos.Address) {
-			defer wg.Done()
-			delegate, err := collector.GetDelegateFromCycle(defaultCtx, cycle, addr)
-			if err != nil {
-				channels <- err
-				return
-			}
+	err = runInParallel(defaultCtx, delegates, constants.OGUN_DELEGATE_FETCH_BATCH_SIZE, func(ctx context.Context, addr tezos.Address, mtx *sync.RWMutex) bool {
+		delegate, err := collector.GetDelegateFromCycle(defaultCtx, cycle, addr)
+		if err != nil {
+			assert.Nil(err)
+			return true
+		}
 
-			_, err = collector.GetDelegationState(defaultCtx, delegate, 745)
-			if err != nil && err != constants.ErrDelegateHasNoMinimumDelegatedBalance {
-				channels <- err
-				return
-			}
-			channels <- nil
-		}(addr)
-	}
-	wg.Wait()
-	for i := 0; i < len(delegates); i++ {
-		if err := <-channels; err != nil {
-			fmt.Println(delegates[i].String())
+		_, err = collector.GetDelegationState(defaultCtx, delegate, 745)
+		if err != nil && err != constants.ErrDelegateHasNoMinimumDelegatedBalance {
+			assert.Nil(err)
+			return true
 		}
-		assert.Nil(<-channels)
-	}
+		return false
+	})
+	assert.Nil(err)
 }