Fix fake minimum capacity (#1982)

* stash it

* remove comment

internal/component/prometheus/write/queue/e2e_stats_test.go

@@ -397,8 +397,9 @@ func TestMetrics(t *testing.T) {
 				},
 			},
 		},
-		// exemplar, note that once it hits the appender exemplars are treated the same as series.
-		{
+		// TURNING OFF EXEMPLAR TESTS until underlying issue is resolved.
+		//exemplar, note that once it hits the appender exemplars are treated the same as series.
+		/*{
 			name:             "exemplar success",
 			returnStatusCode: http.StatusOK,
 			dtype:            Exemplar,
@@ -521,7 +522,7 @@ func TestMetrics(t *testing.T) {
 					valueFunc: isReasonableTimeStamp,
 				},
 			},
-		},
+		},*/
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {

internal/component/prometheus/write/queue/endpoint.go

@@ -31,7 +31,7 @@ func NewEndpoint(client types.NetworkClient, serializer types.Serializer, ttl ti
 		serializer: serializer,
 		log:        logger,
 		ttl:        ttl,
-		incoming:   actor.NewMailbox[types.DataHandle](actor.OptCapacity(1)),
+		incoming:   actor.NewMailbox[types.DataHandle](),
 		buf:        make([]byte, 0, 1024),
 	}
 }

internal/component/prometheus/write/queue/fanout.go

@@ -45,13 +45,17 @@ func (f fanout) Rollback() error {
 }
 
 func (f fanout) AppendExemplar(ref storage.SeriesRef, l labels.Labels, e exemplar.Exemplar) (storage.SeriesRef, error) {
-	for _, child := range f.children {
-		_, err := child.AppendExemplar(ref, l, e)
-		if err != nil {
-			return ref, err
-		}
-	}
+	// Exemplars are disabled due to https://github.com/grafana/alloy/issues/1915
 	return ref, nil
+	/*
+		for _, child := range f.children {
+			_, err := child.AppendExemplar(ref, l, e)
+			if err != nil {
+				return ref, err
+			}
+		}
+		return ref, nil
+	*/
 }
 
 func (f fanout) AppendHistogram(ref storage.SeriesRef, l labels.Labels, t int64, h *histogram.Histogram, fh *histogram.FloatHistogram) (storage.SeriesRef, error) {

internal/component/prometheus/write/queue/network/loop.go

@@ -49,7 +49,7 @@ func newLoop(cc types.ConnectionConfig, isMetaData bool, l log.Logger, stats fun
 	return &loop{
 		isMeta: isMetaData,
 		// In general we want a healthy queue of items, in this case we want to have 2x our maximum send sized ready.
-		seriesMbx:      actor.NewMailbox[*types.TimeSeriesBinary](actor.OptCapacity(2 * cc.BatchCount)),
+		seriesMbx:      actor.NewMailbox[*types.TimeSeriesBinary](actor.OptCapacity(2*cc.BatchCount), actor.OptAsChan()),
 		client:         &http.Client{},
 		cfg:            cc,
 		log:            log.With(l, "name", "loop", "url", cc.URL),

internal/component/prometheus/write/queue/network/manager.go

@@ -37,9 +37,9 @@ func New(cc types.ConnectionConfig, logger log.Logger, seriesStats, metadataStat
 		loops:  make([]*loop, 0, cc.Connections),
 		logger: logger,
 		// This provides blocking to only handle one at a time, so that if a queue blocks
-		// it will stop the filequeue from feeding more.
-		inbox:       actor.NewMailbox[*types.TimeSeriesBinary](actor.OptCapacity(1)),
-		metaInbox:   actor.NewMailbox[*types.TimeSeriesBinary](actor.OptCapacity(1)),
+		// it will stop the filequeue from feeding more. Without OptAsChan the minimum capacity is actually a 64-item buffer.
+		inbox:       actor.NewMailbox[*types.TimeSeriesBinary](actor.OptCapacity(1), actor.OptAsChan()),
+		metaInbox:   actor.NewMailbox[*types.TimeSeriesBinary](actor.OptCapacity(1), actor.OptAsChan()),
 		configInbox: actor.NewMailbox[configCallback](),
 		stats:       seriesStats,
 		metaStats:   metadataStats,
@@ -126,8 +126,8 @@ func (s *manager) DoWork(ctx actor.Context) actor.WorkerStatus {
 			level.Error(s.logger).Log("msg", "failed to send to metadata loop", "err", err)
 		}
 		return actor.WorkerContinue
-		// We need to also check the config here, else its possible this will deadlock.
 	case cfg, ok := <-s.configInbox.ReceiveC():
+		// We need to also check the config here, else its possible this will deadlock.
 		if !ok {
 			level.Debug(s.logger).Log("msg", "config inbox closed")
 			return actor.WorkerEnd

internal/component/prometheus/write/queue/network/manager_test.go

@@ -135,10 +135,8 @@ func TestRetry(t *testing.T) {
 	require.NoError(t, err)
 	wr.Start()
 	defer wr.Stop()
+	send(t, wr, ctx)
 
-	for i := 0; i < 10; i++ {
-		send(t, wr, ctx)
-	}
 	require.Eventually(t, func() bool {
 		done := retries.Load() > 5
 		return done

internal/component/prometheus/write/queue/types/network.go

@@ -10,7 +10,9 @@ import (
 type NetworkClient interface {
 	Start()
 	Stop()
+	// SendSeries will block if the network caches are full.
 	SendSeries(ctx context.Context, d *TimeSeriesBinary) error
+	// SendMetadata will block if the network caches are full.
 	SendMetadata(ctx context.Context, d *TimeSeriesBinary) error
 	// UpdateConfig is a synchronous call and will only return once the config
 	// is applied or an error occurs.