when que overflowed: wait a bit and stop accepting, return errors on …

…feed

utils/queue.go

@@ -4,16 +4,18 @@ import (
 	"context"
 	"errors"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
 type FDQueue[S ~[]E, E any] struct {
-	ctx       context.Context
-	close     context.CancelFunc
-	timelimit time.Duration
-	ch        chan E
-	active    sync.WaitGroup
-	batchSize int
+	ctx        context.Context
+	close      context.CancelFunc
+	timelimit  time.Duration
+	ch         chan E
+	active     sync.WaitGroup
+	batchSize  int
+	overflowed atomic.Bool
 }
 
 var ErrClosed = errors.New("[chotki] feed/drain queue is closed")
@@ -38,7 +40,7 @@ func (q *FDQueue[S, E]) Close() error {
 }
 
 func (q *FDQueue[S, E]) Drain(ctx context.Context, recs S) error {
-	if q.ctx.Err() != nil {
+	if q.ctx.Err() != nil || q.overflowed.Load() {
 		return ErrClosed
 	}
 	q.active.Add(1)
@@ -49,6 +51,8 @@ func (q *FDQueue[S, E]) Drain(ctx context.Context, recs S) error {
 			break
 		case <-q.ctx.Done():
 			break
+		case <-time.After(q.timelimit):
+			q.overflowed.Store(true)
 		case q.ch <- pkg:
 		}
 
@@ -57,7 +61,7 @@ func (q *FDQueue[S, E]) Drain(ctx context.Context, recs S) error {
 }
 
 func (q *FDQueue[S, E]) Feed(ctx context.Context) (recs S, err error) {
-	if q.ctx.Err() != nil {
+	if q.ctx.Err() != nil || q.overflowed.Load() {
 		return nil, ErrClosed
 	}
 	q.active.Add(1)