queue update

Author: GrePod

client/attestation/attestation.go

@@ -13,7 +13,6 @@ import (
 	"github.com/flare-foundation/go-flare-common/pkg/database"
 	"github.com/flare-foundation/go-flare-common/pkg/events"
 	"github.com/flare-foundation/go-flare-common/pkg/logger"
-	"github.com/flare-foundation/go-flare-common/pkg/queue"
 
 	bitvotes "github.com/flare-foundation/fdc-client/client/attestation/bitVotes"
 	"github.com/flare-foundation/fdc-client/client/config"
@@ -129,15 +128,23 @@ func AttestationFromDatabaseLog(request database.Log) (Attestation, error) {
 
 // Handle sends the attestation request to the correct verifier server and validates the response.
 // The response is saved in the struct.
-func (a *Attestation) Handle(ctx context.Context) error {
-	if a.Status == Success || *a.RoundStatus == Done {
-		logger.Debugf("discarding request in round %d", a.RoundID)
-		return fmt.Errorf("%s, handling already confirmed request or round closed", queue.NotRatedDequeue)
+func (a *Attestation) Discard(ctx context.Context) bool {
+	if a.Status == Success {
+		logger.Debugf("discarding already confirmed request in round %d", a.RoundID)
+		return true
+	} else if *a.RoundStatus == Done {
+		logger.Debugf("discarding request in finished round %d", a.RoundID)
+		return true
 	} else if *a.RoundStatus == Consensus && !a.Consensus {
-		logger.Debugf("discarding request in round %d", a.RoundID)
-		return fmt.Errorf("%s, delayed request not in consensus", queue.NotRatedDequeue)
+		logger.Debugf("discarding unselected request in round %d", a.RoundID)
+		return true
 	}
+	return false
+}
 
+// Handle sends the attestation request to the correct verifier server and validates the response.
+// The response is saved in the struct.
+func (a *Attestation) Handle(ctx context.Context) error {
 	responseBytes, confirmed, err := ResolveAttestationRequest(ctx, a)
 	if err != nil {
 		a.Status = ProcessError

client/manager/queues.go

@@ -2,7 +2,6 @@ package manager
 
 import (
 	"context"
-	"time"
 
 	"github.com/flare-foundation/go-flare-common/pkg/logger"
 	"github.com/flare-foundation/go-flare-common/pkg/queue"
@@ -35,6 +34,11 @@ func handler(ctx context.Context, at *attestation.Attestation) error {
 	return at.Handle(ctx)
 }
 
+// discard discards requests that do not need to be handled
+func discard(ctx context.Context, at *attestation.Attestation) bool {
+	return at.Discard(ctx)
+}
+
 // runQueues runs all attestation queues at once.
 func runQueues(ctx context.Context, queues attestationQueues) {
 	for k := range queues {
@@ -46,29 +50,15 @@ func runQueues(ctx context.Context, queues attestationQueues) {
 
 // run tracks and handles all dequeued attestations from a queue.
 func run(ctx context.Context, queue *attestationQueue) {
-	stop := make(chan error)
-
 	for {
-		select {
-		case err := <-stop:
+		err := queue.DequeueAsync(ctx, handler, discard)
+		if err != nil {
+			logger.Warn(err)
+		}
+
+		if err := ctx.Err(); err != nil {
 			logger.Infof("queue worker exiting: %v", err)
 			return
-
-		default:
-			if queue.Length() > 0 {
-				go func() {
-					err := queue.Dequeue(ctx, handler)
-					if err != nil {
-						logger.Warn(err)
-					}
-
-					if err := ctx.Err(); err != nil {
-						stop <- err
-					}
-				}()
-			} else {
-				time.Sleep(10 * time.Millisecond)
-			}
 		}
 	}
 }

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/BurntSushi/toml v1.4.0
 	github.com/bradleyjkemp/cupaloy v2.3.0+incompatible
 	github.com/ethereum/go-ethereum v1.14.11
+	github.com/flare-foundation/go-flare-common v1.1.0-rc.0
 	github.com/gorilla/mux v1.8.1
 	github.com/pkg/errors v0.9.1
 	github.com/rs/cors v1.11.1
@@ -33,7 +34,6 @@ require (
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 // indirect
 	github.com/ethereum/c-kzg-4844 v1.0.3 // indirect
 	github.com/ethereum/go-verkle v0.1.1-0.20240829091221-dffa7562dbe9 // indirect
-	github.com/flare-foundation/go-flare-common v1.0.3-0.20250121164826-d04e4a8b2979 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.6 // indirect
 	github.com/gballet/go-libpcsclite v0.0.0-20191108122812-4678299bea08 // indirect
@@ -83,6 +83,7 @@ require (
 	golang.org/x/sync v0.8.0 // indirect
 	golang.org/x/sys v0.26.0 // indirect
 	golang.org/x/text v0.19.0 // indirect
+	golang.org/x/time v0.5.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

go.sum

@@ -72,16 +72,8 @@ github.com/ethereum/go-ethereum v1.14.11 h1:8nFDCUUE67rPc6AKxFj7JKaOa2W/W1Rse3oS
 github.com/ethereum/go-ethereum v1.14.11/go.mod h1:+l/fr42Mma+xBnhefL/+z11/hcmJ2egl+ScIVPjhc7E=
 github.com/ethereum/go-verkle v0.1.1-0.20240829091221-dffa7562dbe9 h1:8NfxH2iXvJ60YRB8ChToFTUzl8awsc3cJ8CbLjGIl/A=
 github.com/ethereum/go-verkle v0.1.1-0.20240829091221-dffa7562dbe9/go.mod h1:M3b90YRnzqKyyzBEWJGqj8Qff4IDeXnzFw0P9bFw3uk=
-github.com/flare-foundation/go-flare-common v1.0.1 h1:BaOCm8YhZ4mGOXM0Qxj/rpdHirrVwbxdhXBTL/2mqQU=
-github.com/flare-foundation/go-flare-common v1.0.1/go.mod h1:Pb1OAtNe8jNsD2fh6syJD6kVdqxAAXWESTGgd7uQlME=
-github.com/flare-foundation/go-flare-common v1.0.2 h1:GZTQMmIBMVhEMMe2smuGu46jWNoe6+xzeE68XAhmxSE=
-github.com/flare-foundation/go-flare-common v1.0.2/go.mod h1:Pb1OAtNe8jNsD2fh6syJD6kVdqxAAXWESTGgd7uQlME=
-github.com/flare-foundation/go-flare-common v1.0.3-0.20250121134100-511abd70f402 h1:RdcAT9eptqC7na/Rrx7AX2Hbgsh5d34VyQEEAWUN2hI=
-github.com/flare-foundation/go-flare-common v1.0.3-0.20250121134100-511abd70f402/go.mod h1:Pb1OAtNe8jNsD2fh6syJD6kVdqxAAXWESTGgd7uQlME=
-github.com/flare-foundation/go-flare-common v1.0.3-0.20250121135110-393570c3d381 h1:ht1lL0MQ+NqPygYF23tEZu6DIIhX0ymPoWyszncRmYc=
-github.com/flare-foundation/go-flare-common v1.0.3-0.20250121135110-393570c3d381/go.mod h1:Pb1OAtNe8jNsD2fh6syJD6kVdqxAAXWESTGgd7uQlME=
-github.com/flare-foundation/go-flare-common v1.0.3-0.20250121164826-d04e4a8b2979 h1:0F6pHZRPgUhyTysR3FAHnYSw0WiJwiLW0vIEM04wVys=
-github.com/flare-foundation/go-flare-common v1.0.3-0.20250121164826-d04e4a8b2979/go.mod h1:Pb1OAtNe8jNsD2fh6syJD6kVdqxAAXWESTGgd7uQlME=
+github.com/flare-foundation/go-flare-common v1.1.0-rc.0 h1:Fkf9/RCKHa7Ynxi2G6wm9TAq9AvF3nY7jhYbmrN6BSc=
+github.com/flare-foundation/go-flare-common v1.1.0-rc.0/go.mod h1:znM2IpczjntjZGO4cHJKqrtdwCF1+2wWLXqAo1Ag1ek=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=