From 6093eb5353fd89c506b8ba77d1e2b1338e9e87ac Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 09:25:38 +0200
Subject: [PATCH 01/24] Add fast path when onle one shard is configured

---
 internal/component/common/loki/client/shards.go | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/internal/component/common/loki/client/shards.go b/internal/component/common/loki/client/shards.go
index 85321af8f00..5c5350e1807 100644
--- a/internal/component/common/loki/client/shards.go
+++ b/internal/component/common/loki/client/shards.go
@@ -373,8 +373,13 @@ func (s *shards) enqueue(tenantID string, entry loki.Entry, segmentNum int) bool
 		s.initBatchMetrics(tenantID)
 	}
 
-	fingerprint := entry.Labels.FastFingerprint()
-	shard := uint64(fingerprint) % uint64(len(s.queues))
+	var shard uint64
+	// If only one shard is configurec we don't need to compute FastFingerprint.
+	if len(s.queues) == 1 {
+		shard = 0
+	} else {
+		shard = uint64(entry.Labels.FastFingerprint()) % uint64(len(s.queues))
+	}
 
 	select {
 	case <-s.softShutdown:

From 64e13116aab20a85f621599e5a2d24054ad73762 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 11:06:28 +0200
Subject: [PATCH 02/24] Add tests for stopping consumers when queue is full

---
 .../loki/client/consumer_fanout_test.go       | 48 +++++++++++++
 .../common/loki/client/consumer_wal_test.go   | 68 ++++++++++++++++++-
 .../component/common/loki/client/util_test.go | 52 ++++++++++++++
 3 files changed, 165 insertions(+), 3 deletions(-)
 create mode 100644 internal/component/common/loki/client/util_test.go

diff --git a/internal/component/common/loki/client/consumer_fanout_test.go b/internal/component/common/loki/client/consumer_fanout_test.go
index 5cacf431c3f..c79b0f5df54 100644
--- a/internal/component/common/loki/client/consumer_fanout_test.go
+++ b/internal/component/common/loki/client/consumer_fanout_test.go
@@ -205,3 +205,51 @@ func newServerAndEndpointConfig(t *testing.T) (Config, chan util.RemoteWriteRequ
 		close(receivedReqsChan)
 	}
 }
+
+func TestFanoutConsumer_StopWithFullSendQueue(t *testing.T) {
+	const drainTimeout = time.Second
+
+	server, blocked, release := newBlockedServer()
+	defer server.Close()
+	defer release()
+
+	serverURL, err := url.Parse(server.URL)
+	require.NoError(t, err)
+
+	endpointConfig := Config{
+		Name: "test-client",
+		URL:  flagext.URLValue{URL: serverURL},
+		// Long enough that the in-flight request stays parked for the whole test.
+		Timeout:   time.Minute,
+		BatchSize: 1,
+		BackoffConfig: backoff.Config{
+			MinBackoff: time.Millisecond,
+			MaxBackoff: 10 * time.Millisecond,
+			MaxRetries: 0,
+		},
+		QueueConfig: QueueConfig{
+			Capacity:        1,
+			MinShards:       1,
+			DrainTimeout:    drainTimeout,
+			BlockOnOverflow: true,
+		},
+	}
+
+	consumer, err := NewFanoutConsumer(logging.NewSlogNop(), prometheus.NewRegistry(), endpointConfig)
+	require.NoError(t, err)
+
+	feedUntilBlocked(t, blocked, consumer.Chan())
+
+	done := make(chan struct{})
+	go func() {
+		consumer.Stop()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * drainTimeout):
+		release()
+		t.Fatal("Stop did not finish in time")
+	}
+}
diff --git a/internal/component/common/loki/client/consumer_wal_test.go b/internal/component/common/loki/client/consumer_wal_test.go
index 838b823167c..dc18fb8896e 100644
--- a/internal/component/common/loki/client/consumer_wal_test.go
+++ b/internal/component/common/loki/client/consumer_wal_test.go
@@ -305,7 +305,7 @@ func TestWALEndpoint(t *testing.T) {
 			}
 
 			// Stop the endpoint: it waits until the current batch is sent
-			adapter.Stop()
+			adapter.stop()
 			close(receivedReqsChan)
 		})
 	}
@@ -450,7 +450,7 @@ func runWALEndpointBenchCase(b *testing.B, bc testCase, mhFactory func(t *testin
 	}
 
 	// Stop the endpoint: it waits until the current batch is sent
-	adapter.Stop()
+	adapter.stop()
 	close(receivedReqsChan)
 }
 
@@ -512,7 +512,7 @@ func runEndpointBenchCase(b *testing.B, bc testCase) {
 		// Send all the input log entries
 		for j, l := range lines {
 			seriesId := j % bc.numSeries
-			endpoint.enqueue(loki.Entry{
+			endpoint.enqueue(b.Context(), loki.Entry{
 				Labels: model.LabelSet{
 					// take j module bc.numSeries to evenly distribute those numSeries across all sent entries
 					"app": model.LabelValue(fmt.Sprintf("series-%d", seriesId)),
@@ -536,3 +536,65 @@ func runEndpointBenchCase(b *testing.B, bc testCase) {
 	endpoint.stop()
 	close(receivedReqsChan)
 }
+
+func TestWALConsumer_StopWithFullSendQueue(t *testing.T) {
+	const (
+		drainTimeout = time.Second
+	)
+
+	server, blocked, release := newBlockedServer()
+	defer server.Close()
+	defer release()
+
+	serverURL, err := url.Parse(server.URL)
+	require.NoError(t, err)
+
+	walConfig := wal.Config{
+		Dir:           t.TempDir(),
+		Enabled:       true,
+		MaxSegmentAge: time.Minute,
+		WatchConfig: wal.WatchConfig{
+			MinReadFrequency: 10 * time.Millisecond,
+			MaxReadFrequency: 50 * time.Millisecond,
+			DrainTimeout:     drainTimeout,
+		},
+	}
+
+	endpointConfig := Config{
+		Name: "test-client",
+		URL:  flagext.URLValue{URL: serverURL},
+		// Long enough that the in-flight request stays parked for the whole test.
+		Timeout:   time.Minute,
+		BatchSize: 1,
+		BackoffConfig: backoff.Config{
+			MinBackoff: time.Millisecond,
+			MaxBackoff: 10 * time.Millisecond,
+			MaxRetries: 0,
+		},
+		QueueConfig: QueueConfig{
+			Capacity:        1,
+			MinShards:       1,
+			DrainTimeout:    drainTimeout,
+			BlockOnOverflow: true,
+		},
+	}
+
+	consumer, err := NewWALConsumer(logging.NewSlogNop(), prometheus.NewRegistry(), walConfig, endpointConfig)
+	require.NoError(t, err)
+
+	feedUntilBlocked(t, blocked, consumer.Chan())
+
+	stopped := make(chan struct{})
+	go func() {
+		consumer.StopAndDrain()
+		close(stopped)
+	}()
+
+	// A healthy shutdown costs at most the WAL drain timeout plus the queue drain timeout.
+	select {
+	case <-stopped:
+	case <-time.After(5 * (drainTimeout + drainTimeout)):
+		release()
+		t.Fatal("StopAndDrain did not finish in time")
+	}
+}
diff --git a/internal/component/common/loki/client/util_test.go b/internal/component/common/loki/client/util_test.go
new file mode 100644
index 00000000000..4a6333abaff
--- /dev/null
+++ b/internal/component/common/loki/client/util_test.go
@@ -0,0 +1,52 @@
+package client
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/grafana/loki/pkg/push"
+	"github.com/prometheus/common/model"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+
+	"github.com/grafana/alloy/internal/component/common/loki"
+)
+
+func newBlockedServer() (*httptest.Server, *atomic.Bool, func()) {
+	var (
+		done     = make(chan struct{})
+		doneOnce sync.Once
+		blocked  = atomic.NewBool(false)
+	)
+
+	release := func() { doneOnce.Do(func() { close(done) }) }
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		blocked.Store(true)
+		select {
+		case <-done:
+		case <-r.Context().Done():
+		}
+	}))
+
+	return server, blocked, release
+}
+
+func feedUntilBlocked(t *testing.T, blocked *atomic.Bool, c chan<- loki.Entry) {
+	e := loki.NewEntry(model.LabelSet{"A": "b"}, push.Entry{
+		Line:      "test",
+		Timestamp: time.Now(),
+	})
+
+	for !blocked.Load() {
+		select {
+		case c <- e:
+		case <-time.After(50 * time.Millisecond):
+		}
+	}
+	require.True(t, blocked.Load())
+
+}

From eb9ca9fd6f0a20884a35b07204b143c9da3e5a3b Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 12:43:04 +0200
Subject: [PATCH 03/24] Update enqueue to accept context from caller.

With this caller can cancel any enqueue retries
---
 .../common/loki/client/consumer_fanout.go     |  30 ++++--
 .../common/loki/client/consumer_wal.go        |  17 +--
 .../component/common/loki/client/endpoint.go  |  40 +++----
 .../common/loki/client/endpoint_test.go       | 100 ++++++++++++++++--
 .../component/common/loki/client/shards.go    |   2 +-
 5 files changed, 135 insertions(+), 54 deletions(-)

diff --git a/internal/component/common/loki/client/consumer_fanout.go b/internal/component/common/loki/client/consumer_fanout.go
index 23d820e54a2..d62950d5e8b 100644
--- a/internal/component/common/loki/client/consumer_fanout.go
+++ b/internal/component/common/loki/client/consumer_fanout.go
@@ -1,6 +1,7 @@
 package client
 
 import (
+	"context"
 	"fmt"
 	"log/slog"
 	"sync"
@@ -16,11 +17,13 @@ func NewFanoutConsumer(logger *slog.Logger, reg prometheus.Registerer, cfgs ...C
 		return nil, fmt.Errorf("at least one endpoint config must be provided")
 	}
 
-	m := &FanoutConsumer{
+	c := &FanoutConsumer{
 		endpoints: make([]*endpoint, 0, len(cfgs)),
 		recv:      make(chan loki.Entry),
 	}
 
+	c.ctx, c.cancel = context.WithCancel(context.Background())
+
 	var (
 		metrics        = newMetrics(reg)
 		endpointsCheck = make(map[string]struct{})
@@ -39,27 +42,30 @@ func NewFanoutConsumer(logger *slog.Logger, reg prometheus.Registerer, cfgs ...C
 			return nil, fmt.Errorf("error starting endpoint: %w", err)
 		}
 
-		m.endpoints = append(m.endpoints, endpoint)
+		c.endpoints = append(c.endpoints, endpoint)
 	}
 
-	m.wg.Go(m.run)
-	return m, nil
+	c.wg.Go(c.run)
+	return c, nil
 }
 
 var _ Consumer = (*FanoutConsumer)(nil)
 
 type FanoutConsumer struct {
 	endpoints []*endpoint
-	wg        sync.WaitGroup
-	once      sync.Once
-	recv      chan loki.Entry
+
+	wg     sync.WaitGroup
+	once   sync.Once
+	recv   chan loki.Entry
+	ctx    context.Context
+	cancel context.CancelFunc
 }
 
 func (c *FanoutConsumer) run() {
 	for e := range c.recv {
-		for _, c := range c.endpoints {
+		for _, endpoint := range c.endpoints {
 			// NOTE: For now it's fine to ignore error because we can't act on it.
-			_ = c.enqueue(e, 0)
+			_ = endpoint.enqueue(c.ctx, e, 0)
 		}
 	}
 }
@@ -70,7 +76,11 @@ func (c *FanoutConsumer) Chan() chan<- loki.Entry {
 
 func (c *FanoutConsumer) Stop() {
 	// First stop the receiving channel.
-	c.once.Do(func() { close(c.recv) })
+	c.once.Do(func() {
+		close(c.recv)
+		c.cancel()
+	})
+
 	c.wg.Wait()
 
 	var stopWG sync.WaitGroup
diff --git a/internal/component/common/loki/client/consumer_wal.go b/internal/component/common/loki/client/consumer_wal.go
index 5641c047116..ec7bb4e1acc 100644
--- a/internal/component/common/loki/client/consumer_wal.go
+++ b/internal/component/common/loki/client/consumer_wal.go
@@ -1,6 +1,7 @@
 package client
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -89,8 +90,8 @@ type endpointWatcherPair struct {
 	endpoint *walEndpointAdapter
 }
 
-// Stop will proceed to stop, in order, watcher and the endpoint.
-func (p endpointWatcherPair) Stop(drain bool) {
+// stop will proceed to stop, in order, watcher and the endpoint.
+func (p endpointWatcherPair) stop(drain bool) {
 	// If drain enabled, drain the WAL.
 	if drain {
 		p.watcher.Drain()
@@ -98,7 +99,7 @@ func (p endpointWatcherPair) Stop(drain bool) {
 	p.watcher.Stop()
 
 	// Subsequently stop the endpoint.
-	p.endpoint.Stop()
+	p.endpoint.stop()
 }
 
 var _ DrainableConsumer = (*WALConsumer)(nil)
@@ -132,7 +133,7 @@ func (m *WALConsumer) stop(drain bool) {
 	// endpoint config, each (watcher, queue) pair is stopped concurrently.
 	for _, pair := range m.pairs {
 		stopWG.Go(func() {
-			pair.Stop(drain)
+			pair.stop(drain)
 		})
 	}
 
@@ -206,7 +207,7 @@ func (c *walEndpointAdapter) AppendEntries(entries wal.RefEntries, segment int)
 	if ok {
 		for i := range entries.Entries {
 			e := entries.EntryAt(l, i)
-			err := c.endpoint.enqueue(e, segment)
+			err := c.endpoint.enqueue(context.Background(), e, segment)
 			// We can receive errQueueIsFull if we have configured endpoint with BlockOnOverflow.
 			// Here we just skip the entry and try with the next one.
 			if errors.Is(err, errQueueIsFull) {
@@ -236,9 +237,9 @@ func (c *walEndpointAdapter) AppendEntries(entries wal.RefEntries, segment int)
 	return nil
 }
 
-// Stop the endpoint, enqueueing pending batches and draining the send queue accordingly. Both closing operations are
-// limited by a deadline, controlled by a configured drain timeout, which is global to the Stop call.
-func (c *walEndpointAdapter) Stop() {
+// stop the endpoint, enqueueing pending batches and draining the send queue accordingly. Both closing operations are
+// limited by a deadline, controlled by a configured drain timeout, which is global to the stop call.
+func (c *walEndpointAdapter) stop() {
 	c.endpoint.stop()
 	c.tracker.Stop()
 }
diff --git a/internal/component/common/loki/client/endpoint.go b/internal/component/common/loki/client/endpoint.go
index c7e116f0501..c37b40d4d7a 100644
--- a/internal/component/common/loki/client/endpoint.go
+++ b/internal/component/common/loki/client/endpoint.go
@@ -18,13 +18,7 @@ type endpoint struct {
 	cfg     Config
 	metrics *metrics
 	logger  *slog.Logger
-	entries chan loki.Entry
-
-	ctx    context.Context
-	cancel context.CancelFunc
-
 	shards  *shards
-	backoff *backoff.Backoff
 }
 
 func newEndpoint(metrics *metrics, cfg Config, logger *slog.Logger, markerHandler marker.Tracker) (*endpoint, error) {
@@ -35,19 +29,11 @@ func newEndpoint(metrics *metrics, cfg Config, logger *slog.Logger, markerHandle
 		return nil, err
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
 	c := &endpoint{
 		cfg:     cfg,
 		logger:  logger,
 		metrics: metrics,
-		entries: make(chan loki.Entry),
-		ctx:     ctx,
-		cancel:  cancel,
 		shards:  shards,
-		backoff: backoff.New(ctx, backoff.Config{
-			MinBackoff: 5 * time.Millisecond,
-			MaxBackoff: 50 * time.Millisecond,
-		}),
 	}
 
 	c.shards.start(cfg.QueueConfig.MinShards)
@@ -57,30 +43,34 @@ func newEndpoint(metrics *metrics, cfg Config, logger *slog.Logger, markerHandle
 var errQueueIsFull = errors.New("queue is full")
 
 // enqueue tries to enqueue an entry. It returns an error if the entry could not be enqueued.
-// errQueueIsFull when the queue is full and BlockOnOverflow is false, or context.Canceled when
-// endpoint is stopped.
-func (e *endpoint) enqueue(entry loki.Entry, segmentNum int) error {
-	defer e.backoff.Reset()
+// errQueueIsFull when the queue is full and BlockOnOverflow is false, or context.Canceled if
+// caller canceled ctx.
+func (e *endpoint) enqueue(ctx context.Context, entry loki.Entry, segmentNum int) error {
+	bo := backoff.New(ctx, backoff.Config{
+		MinBackoff: 5 * time.Millisecond,
+		MaxBackoff: 50 * time.Millisecond,
+	})
 
 	tenantID := getTenantID(e.cfg, entry)
-	for !e.shards.enqueue(tenantID, entry, segmentNum) {
+
+	for bo.Ongoing() {
+		if e.shards.enqueue(tenantID, entry, segmentNum) {
+			return nil
+		}
+
 		if !e.cfg.QueueConfig.BlockOnOverflow {
 			e.metrics.droppedEntries.WithLabelValues(e.cfg.URL.Host, tenantID, reasonQueueIsFull).Inc()
 			e.metrics.droppedBytes.WithLabelValues(e.cfg.URL.Host, tenantID, reasonQueueIsFull).Add(float64(entry.Size()))
 			return errQueueIsFull
 		}
 
-		e.backoff.Wait()
-		if !e.backoff.Ongoing() {
-			return e.backoff.Err()
-		}
+		bo.Wait()
 	}
 
-	return nil
+	return bo.Err()
 }
 
 func (e *endpoint) stop() {
-	e.cancel()
 	e.shards.stop()
 }
 
diff --git a/internal/component/common/loki/client/endpoint_test.go b/internal/component/common/loki/client/endpoint_test.go
index e1b2d1b2ab7..49c8c8cfa19 100644
--- a/internal/component/common/loki/client/endpoint_test.go
+++ b/internal/component/common/loki/client/endpoint_test.go
@@ -1,8 +1,10 @@
 package client
 
 import (
+	"context"
 	"errors"
 	"net/http"
+	"net/http/httptest"
 	"runtime"
 	"strings"
 	"testing"
@@ -17,6 +19,7 @@ import (
 	"github.com/prometheus/common/config"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
 
 	"github.com/grafana/alloy/internal/component/common/loki"
 	"github.com/grafana/alloy/internal/component/common/loki/client/internal/marker"
@@ -341,7 +344,7 @@ func TestEndpoint(t *testing.T) {
 
 			// Send all the input log entries
 			for i, logEntry := range tt.inputEntries {
-				c.enqueue(logEntry, 0)
+				c.enqueue(t.Context(), logEntry, 0)
 
 				if tt.inputDelay > 0 && i < len(tt.inputEntries)-1 {
 					time.Sleep(tt.inputDelay)
@@ -402,14 +405,14 @@ func TestEndpointBlockOnOverflow(t *testing.T) {
 		// NOTE: We have configured batch size to 1 so only one entry will fit in each batch.
 		// To exceed the queue's capacity we need to pass 4 entries. We have one batch that we are actively trying
 		// to send, one batch that is queued and one batch that we are currently working with filling up.
-		require.NoError(t, e.enqueue(entry, 0))
-		require.NoError(t, e.enqueue(entry, 0))
+		require.NoError(t, e.enqueue(t.Context(), entry, 0))
+		require.NoError(t, e.enqueue(t.Context(), entry, 0))
 
 		// Which enqueue fails depends on whether the shard worker has already
 		// consumed the queued batch after the third call. If the third call loses that race,
 		// it returns errQueueIsFull, otherwise the fourth call does.
-		err3 := e.enqueue(entry, 0)
-		err4 := e.enqueue(entry, 0)
+		err3 := e.enqueue(t.Context(), entry, 0)
+		err4 := e.enqueue(t.Context(), entry, 0)
 		queueIsFull := errors.Is(err3, errQueueIsFull) || errors.Is(err4, errQueueIsFull)
 		require.True(t, queueIsFull, "expected either the third or fourth enqueue to fail with queue full")
 	})
@@ -448,10 +451,10 @@ func TestEndpointBlockOnOverflow(t *testing.T) {
 			// We just need to finish one request in order for all entries to be successfully enqueued.
 			<-receivedReqsChan
 		}()
-		require.NoError(t, e.enqueue(entry1, 0))
-		require.NoError(t, e.enqueue(entry2, 0))
-		require.NoError(t, e.enqueue(entry3, 0))
-		require.NoError(t, e.enqueue(entry4, 0))
+		require.NoError(t, e.enqueue(t.Context(), entry1, 0))
+		require.NoError(t, e.enqueue(t.Context(), entry2, 0))
+		require.NoError(t, e.enqueue(t.Context(), entry3, 0))
+		require.NoError(t, e.enqueue(t.Context(), entry4, 0))
 	})
 }
 
@@ -483,7 +486,7 @@ func TestEndpointBatchSizeMetric(t *testing.T) {
 	require.NoError(t, err)
 
 	for _, entry := range entries {
-		require.NoError(t, e.enqueue(entry, 0))
+		require.NoError(t, e.enqueue(t.Context(), entry, 0))
 	}
 
 	// Stopping the endpoint waits until the current batch is sent.
@@ -498,6 +501,83 @@ func TestEndpointBatchSizeMetric(t *testing.T) {
 	assert.Equal(t, float64(entries[0].Size()+entries[1].Size()), sum)
 }
 
+func TestEndpointCallerCancel(t *testing.T) {
+	t.Run("entry is not queued or sent if callers context is canceled", func(t *testing.T) {
+		called := atomic.NewBool(false)
+
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			called.Store(true)
+		}))
+		defer server.Close()
+
+		var url flagext.URLValue
+		require.NoError(t, url.Set(server.URL))
+
+		e, err := newEndpoint(newMetrics(prometheus.NewRegistry()), Config{URL: url}, logging.NewSlogNop(), marker.NewNopTracker())
+		require.NoError(t, err)
+		defer e.stop()
+
+		ctx, cancel := context.WithCancel(t.Context())
+		cancel()
+
+		require.ErrorIs(t, e.enqueue(ctx, loki.Entry{Entry: push.Entry{Line: "my entry"}}, 0), context.Canceled)
+		require.Equal(t, false, called.Load())
+	})
+
+	t.Run("when queue is full and callers context is done", func(t *testing.T) {
+		server, blocked, release := newBlockedServer()
+		defer server.Close()
+		defer release()
+
+		var url flagext.URLValue
+		require.NoError(t, url.Set(server.URL))
+
+		e, err := newEndpoint(newMetrics(prometheus.NewRegistry()), Config{
+			Name:      "test-client",
+			URL:       url,
+			Timeout:   time.Minute,
+			BatchSize: 1,
+			BackoffConfig: backoff.Config{
+				MinBackoff: time.Millisecond,
+				MaxBackoff: 10 * time.Millisecond,
+				MaxRetries: 0,
+			},
+			QueueConfig: QueueConfig{
+				Capacity:        1,
+				MinShards:       1,
+				DrainTimeout:    1 * time.Second,
+				BlockOnOverflow: true,
+			},
+		}, logging.NewSlogNop(), marker.NewNopTracker())
+		require.NoError(t, err)
+
+		defer e.stop()
+
+		entry := loki.Entry{Entry: push.Entry{Line: "my entry"}}
+
+		require.NoError(t, e.enqueue(t.Context(), entry, 0))
+		require.Eventually(t, func() bool { return blocked.Load() }, 3*time.Second, 100*time.Millisecond)
+
+		require.NoError(t, e.enqueue(t.Context(), entry, 0))
+		require.NoError(t, e.enqueue(t.Context(), entry, 0))
+
+		ctx, cancel := context.WithTimeout(t.Context(), 1*time.Second)
+		defer cancel()
+
+		done := make(chan error, 1)
+		go func() {
+			done <- e.enqueue(ctx, entry, 0)
+		}()
+
+		select {
+		case err := <-done:
+			require.ErrorIs(t, err, context.DeadlineExceeded)
+		case <-time.After(2 * time.Second):
+			t.Fatal("entry trying to be queued was not canceled in time")
+		}
+	})
+}
+
 // histogramSumAndCount returns the sum and count of the single series of the
 // named histogram in reg.
 func histogramSumAndCount(t *testing.T, reg *prometheus.Registry, name string) (float64, uint64) {
diff --git a/internal/component/common/loki/client/shards.go b/internal/component/common/loki/client/shards.go
index 5c5350e1807..b9d49d93380 100644
--- a/internal/component/common/loki/client/shards.go
+++ b/internal/component/common/loki/client/shards.go
@@ -374,7 +374,7 @@ func (s *shards) enqueue(tenantID string, entry loki.Entry, segmentNum int) bool
 	}
 
 	var shard uint64
-	// If only one shard is configurec we don't need to compute FastFingerprint.
+	// If only one shard is configured we don't need to compute FastFingerprint.
 	if len(s.queues) == 1 {
 		shard = 0
 	} else {

From a9b7f9fcae6085a9c12cb56189170666e5b0bc0e Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 13:38:25 +0200
Subject: [PATCH 04/24] Refactor watcher state and thread the stopping context
 through so that blocked enqueues are canceled.

---
 .../common/loki/client/consumer_wal.go        |  9 ++---
 .../common/loki/client/consumer_wal_test.go   |  4 +-
 .../common/loki/wal/internal/watcher_state.go | 32 ++++++++-------
 internal/component/common/loki/wal/watcher.go | 39 +++++++++++++++----
 .../component/common/loki/wal/watcher_test.go |  5 ++-
 5 files changed, 60 insertions(+), 29 deletions(-)

diff --git a/internal/component/common/loki/client/consumer_wal.go b/internal/component/common/loki/client/consumer_wal.go
index ec7bb4e1acc..4efd534be7f 100644
--- a/internal/component/common/loki/client/consumer_wal.go
+++ b/internal/component/common/loki/client/consumer_wal.go
@@ -194,7 +194,7 @@ func (c *walEndpointAdapter) StoreSeries(series []record.RefSeries, segment int)
 	}
 }
 
-func (c *walEndpointAdapter) AppendEntries(entries wal.RefEntries, segment int) error {
+func (c *walEndpointAdapter) AppendEntries(ctx context.Context, entries wal.RefEntries, segment int) error {
 	c.seriesLock.RLock()
 	l, ok := c.series[entries.Ref]
 	c.seriesLock.RUnlock()
@@ -207,16 +207,15 @@ func (c *walEndpointAdapter) AppendEntries(entries wal.RefEntries, segment int)
 	if ok {
 		for i := range entries.Entries {
 			e := entries.EntryAt(l, i)
-			err := c.endpoint.enqueue(context.Background(), e, segment)
+			err := c.endpoint.enqueue(ctx, e, segment)
 			// We can receive errQueueIsFull if we have configured endpoint with BlockOnOverflow.
 			// Here we just skip the entry and try with the next one.
 			if errors.Is(err, errQueueIsFull) {
 				continue
 			}
-			// NOTE: The only other error that can be returned is context.Canceled and that happens
-			// if endpoint was stopped.
+
 			if err != nil {
-				return nil
+				return err
 			}
 
 			queuedEntries += 1
diff --git a/internal/component/common/loki/client/consumer_wal_test.go b/internal/component/common/loki/client/consumer_wal_test.go
index dc18fb8896e..e5d0f473d4c 100644
--- a/internal/component/common/loki/client/consumer_wal_test.go
+++ b/internal/component/common/loki/client/consumer_wal_test.go
@@ -286,7 +286,7 @@ func TestWALEndpoint(t *testing.T) {
 					},
 				}, 0)
 
-				_ = adapter.AppendEntries(wal.RefEntries{
+				_ = adapter.AppendEntries(t.Context(), wal.RefEntries{
 					Ref:     chunks.HeadSeriesRef(mod),
 					Created: time.Now().UnixMicro(),
 					Entries: []push.Entry{{
@@ -431,7 +431,7 @@ func runWALEndpointBenchCase(b *testing.B, bc testCase, mhFactory func(t *testin
 				},
 			}, 0)
 
-			_ = adapter.AppendEntries(wal.RefEntries{
+			_ = adapter.AppendEntries(b.Context(), wal.RefEntries{
 				Ref:     chunks.HeadSeriesRef(seriesId),
 				Created: time.Now().UnixMicro(),
 				Entries: []push.Entry{{
diff --git a/internal/component/common/loki/wal/internal/watcher_state.go b/internal/component/common/loki/wal/internal/watcher_state.go
index b416a094445..56d64486b45 100644
--- a/internal/component/common/loki/wal/internal/watcher_state.go
+++ b/internal/component/common/loki/wal/internal/watcher_state.go
@@ -1,6 +1,7 @@
 package internal
 
 import (
+	"context"
 	"log/slog"
 	"sync"
 )
@@ -21,17 +22,23 @@ const (
 // WatcherState is a holder for the state the Watcher is in. It provides handy methods for checking it it's stopping, getting
 // the current state, or blocking until it has stopped.
 type WatcherState struct {
-	current        int
-	mut            sync.RWMutex
-	stoppingSignal chan struct{}
-	logger         *slog.Logger
+	current int
+	mut     sync.RWMutex
+	logger  *slog.Logger
+
+	// ctx is canceled when the state transitions to StateStopping.
+	ctx    context.Context
+	cancel context.CancelFunc
 }
 
 func NewWatcherState(logger *slog.Logger) *WatcherState {
+	ctx, cancel := context.WithCancel(context.Background())
+
 	return &WatcherState{
-		current:        StateRunning,
-		stoppingSignal: make(chan struct{}),
-		logger:         logger,
+		current: StateRunning,
+		logger:  logger,
+		ctx:     ctx,
+		cancel:  cancel,
 	}
 }
 
@@ -42,10 +49,9 @@ func (s *WatcherState) Transition(next int) {
 
 	s.logger.Debug("watcher transitioning state", "currentState", printState(s.current), "nextState", printState(next))
 
-	// only perform channel close if the state is not already stopping
-	// expect s.s to be either draining ro running to perform a close
+	// only cancel context if the state is not already StateStopping.
 	if next == StateStopping && s.current != next {
-		close(s.stoppingSignal)
+		s.cancel()
 	}
 
 	// update state
@@ -66,9 +72,9 @@ func (s *WatcherState) IsStopping() bool {
 	return s.current == StateStopping
 }
 
-// WaitForStopping returns a channel in which the called can read, effectively waiting until the state changes to stopping.
-func (s *WatcherState) WaitForStopping() <-chan struct{} {
-	return s.stoppingSignal
+// StoppingContext returns a context that will be canceled when state is transitioned to StateStopping.
+func (s *WatcherState) StoppingContext() context.Context {
+	return s.ctx
 }
 
 // printState prints a user-friendly name of the possible Watcher states.
diff --git a/internal/component/common/loki/wal/watcher.go b/internal/component/common/loki/wal/watcher.go
index 515218ed152..86a27d2ba91 100644
--- a/internal/component/common/loki/wal/watcher.go
+++ b/internal/component/common/loki/wal/watcher.go
@@ -1,6 +1,7 @@
 package wal
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -53,7 +54,9 @@ type WriteTo interface {
 	// found in.
 	StoreSeries(series []record.RefSeries, segmentNum int)
 
-	AppendEntries(entries RefEntries, segmentNum int) error
+	// AppendEntries is called with the entries read from a segment. ctx is tied to the lifetime of the Watcher and is
+	// canceled when it stops, so an implementation that blocks must honour it or it will hold up shutdown.
+	AppendEntries(ctx context.Context, entries RefEntries, segmentNum int) error
 }
 
 // Marker allows the Watcher to start from a specific segment in the WAL.
@@ -132,12 +135,12 @@ func (w *Watcher) mainLoop() {
 		if w.state.IsDraining() && errors.Is(err, os.ErrNotExist) {
 			w.logger.Info("reached non existing segment while draining, assuming end of WAL")
 			// since we've reached the end of the WAL, and the Watcher is draining, promptly transition to stopping state
-			// so the watcher can stoppingSignal early
+			// so the watcher can stop early
 			w.state.Transition(internal.StateStopping)
 		}
 
 		select {
-		case <-w.state.WaitForStopping():
+		case <-w.state.StoppingContext().Done():
 			return
 		case <-time.After(5 * time.Second):
 		}
@@ -222,7 +225,7 @@ func (w *Watcher) watch(segmentNum int, tail bool) error {
 
 	for {
 		select {
-		case <-w.state.WaitForStopping():
+		case <-w.state.StoppingContext().Done():
 			return nil
 
 		case <-segmentTicker.C:
@@ -244,8 +247,14 @@ func (w *Watcher) watch(segmentNum int, tail bool) error {
 
 			// We know that there's either a new segment (last > segmentNum), or we are draining the WAL. Either case, read
 			// the remaining data from the segmentNum and return from `watch` to read the next one.
+
 			_, err = w.readSegment(reader, segmentNum)
 
+			// context.Canceled means the Watcher is stopping.
+			if errors.Is(err, context.Canceled) {
+				return nil
+			}
+
 			// io.EOF error are non-fatal since we are consuming the segment till the end
 			if !errors.Is(err, io.EOF) {
 				return err
@@ -264,6 +273,12 @@ func (w *Watcher) watch(segmentNum int, tail bool) error {
 
 		// read from open segment routine
 		ok, err := w.readSegment(reader, segmentNum)
+
+		// context.Canceled means the Watcher is stopping.
+		if errors.Is(err, context.Canceled) {
+			return nil
+		}
+
 		// Ignore all errors reading to end of segment whilst replaying the WAL. This is because when replaying not the
 		// last segment, we assume that segment is not written anymore (closed), and the call to readSegment will read
 		// to the end of it. If error, log a warning accordingly. After, error or no error, nil is returned so that the
@@ -315,7 +330,7 @@ func (w *Watcher) readSegment(r *wlog.LiveReader, segmentNum int) (bool, error)
 }
 
 // decodeAndDispatch first decodes a WAL record. Upon reading either Series or Entries from the WAL record, call the
-// appropriate callbacks in the writeTo.
+// appropriate function in the writeTo.
 func (w *Watcher) decodeAndDispatch(b []byte, segmentNum int) (bool, error) {
 	var readData bool
 
@@ -333,7 +348,15 @@ func (w *Watcher) decodeAndDispatch(b []byte, segmentNum int) (bool, error) {
 	readData = true
 
 	for _, entries := range rec.RefEntries {
-		if err := w.actions.AppendEntries(entries, segmentNum); err != nil && firstErr == nil {
+		err := w.actions.AppendEntries(w.state.StoppingContext(), entries, segmentNum)
+
+		// The WriteTo gave up because the Watcher is stopping, so there is no point handing it
+		// the rest of the record. The caller turns this into a quiet exit.
+		if errors.Is(err, context.Canceled) {
+			return readData, err
+		}
+
+		if err != nil && firstErr == nil {
 			firstErr = err
 		}
 	}
@@ -351,12 +374,14 @@ func (w *Watcher) Drain() {
 	select {
 	case <-time.NewTimer(w.drainTimeout).C:
 		w.logger.Warn("watcher drain timeout occurred, transitioning to Stopping")
-	case <-w.state.WaitForStopping():
+	case <-w.state.StoppingContext().Done():
 	}
 }
 
 // Stop stops the Watcher, shutting down the main routine.
 func (w *Watcher) Stop() {
+	// Transitioning cancels the state's context, which releases a WriteTo blocked in
+	// AppendEntries so the wait below can return.
 	w.state.Transition(internal.StateStopping)
 
 	// upon calling stop, wait for main mainLoop execution to stop
diff --git a/internal/component/common/loki/wal/watcher_test.go b/internal/component/common/loki/wal/watcher_test.go
index 74f04b72a9c..414c1c2f4cd 100644
--- a/internal/component/common/loki/wal/watcher_test.go
+++ b/internal/component/common/loki/wal/watcher_test.go
@@ -2,6 +2,7 @@ package wal
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
 	"fmt"
 	"hash/crc32"
@@ -40,7 +41,7 @@ func (t *testWriteTo) SeriesReset(segmentNum int) {
 	t.ReceivedSeriesReset = append(t.ReceivedSeriesReset, segmentNum)
 }
 
-func (t *testWriteTo) AppendEntries(entries RefEntries, _ int) error {
+func (t *testWriteTo) AppendEntries(_ context.Context, entries RefEntries, _ int) error {
 	var entry loki.Entry
 	if l, ok := t.series[uint64(entries.Ref)]; ok {
 		entry.Labels = l
@@ -584,7 +585,7 @@ func (s *slowWriteTo) SeriesReset(segmentNum int) {
 func (s *slowWriteTo) StoreSeries(series []record.RefSeries, segmentNum int) {
 }
 
-func (s *slowWriteTo) AppendEntries(entries RefEntries, segmentNum int) error {
+func (s *slowWriteTo) AppendEntries(_ context.Context, entries RefEntries, segmentNum int) error {
 	s.entriesReceived.Add(uint64(len(entries.Entries)))
 	time.Sleep(s.sleepAfterAppendEntries)
 	return nil

From f45bf4aeb9ce78cf578a861ef2e2c69a0928bc5c Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 13:43:30 +0200
Subject: [PATCH 05/24] update docs

---
 docs/sources/reference/components/loki/loki.write.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/sources/reference/components/loki/loki.write.md b/docs/sources/reference/components/loki/loki.write.md
index 0bb44414711..cb1e9090ff0 100644
--- a/docs/sources/reference/components/loki/loki.write.md
+++ b/docs/sources/reference/components/loki/loki.write.md
@@ -81,7 +81,7 @@ The following arguments are supported:
 | `http_headers`           | `map(list(secret))` | Custom HTTP headers to be sent along with each request. The map key is the header name.          |           | no       |
 | `headers`                | `map(string)`       | Extra headers to deliver with the request.                                                       |           | no       |
 | `max_backoff_period`     | `duration`          | Maximum backoff time between retries.                                                            | `"5m"`    | no       |
-| `max_backoff_retries`    | `int`               | Maximum number of retries.                                                                       | `10`      | no       |
+| `max_backoff_retries`    | `int`               | Maximum number of retries. Set to `0` to retry indefinitely.                                     | `10`      | no       |
 | `min_backoff_period`     | `duration`          | Initial backoff time between retries.                                                            | `"500ms"` | no       |
 | `name`                   | `string`            | Optional name to identify this endpoint with.                                                    |           | no       |
 | `no_proxy`               | `string`            | Comma-separated list of IP addresses, CIDR notations, and domain names to exclude from proxying. |           | no       |
@@ -139,7 +139,7 @@ The following arguments are supported:
 | -------------------- | ---------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------- | -------- |
 | `block_on_overflow`  | `bool`     | If `true`, block until there is space in the queue; if `false`, drop entries when queue is full.                                                                              | `true`  | no       |
 | `capacity`           | `string`   | Controls the size of the underlying send queue buffer. This setting should be considered a worst-case scenario of memory consumption, in which all enqueued batches are full. | `10MiB` | no       |
-| `drain_timeout`      | `duration` | Configures the maximum time the client can take to drain the send queue upon shutdown. During that time, it enqueues pending batches and drains the send queue sending each.  | `"1m"`  | no       |
+| `drain_timeout`      | `duration` | Configures the maximum time the client can take to drain the send queue upon shutdown. During that time, it enqueues pending batches and drains the send queue sending each.  | `"15s"` | no       |
 | `min_shards`         | `number`   | Minimum number of concurrent shards sending samples to the endpoint.                                                                                                          | `1`     | no       |
 
 Each endpoint is divided into a number of concurrent _shards_ which are responsible for sending a fraction of batches. The number of shards is controlled with `min_shards` argument.

From 424c08d3df95c6786fd1659b8965975280c0da58 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 13:50:17 +0200
Subject: [PATCH 06/24] restore change

---
 internal/component/common/loki/client/shards.go | 9 ++-------
 1 file changed, 2 insertions(+), 7 deletions(-)

diff --git a/internal/component/common/loki/client/shards.go b/internal/component/common/loki/client/shards.go
index b9d49d93380..85321af8f00 100644
--- a/internal/component/common/loki/client/shards.go
+++ b/internal/component/common/loki/client/shards.go
@@ -373,13 +373,8 @@ func (s *shards) enqueue(tenantID string, entry loki.Entry, segmentNum int) bool
 		s.initBatchMetrics(tenantID)
 	}
 
-	var shard uint64
-	// If only one shard is configured we don't need to compute FastFingerprint.
-	if len(s.queues) == 1 {
-		shard = 0
-	} else {
-		shard = uint64(entry.Labels.FastFingerprint()) % uint64(len(s.queues))
-	}
+	fingerprint := entry.Labels.FastFingerprint()
+	shard := uint64(fingerprint) % uint64(len(s.queues))
 
 	select {
 	case <-s.softShutdown:

From d33eceb09ca69580c2c8ca9f478de44cbf5dc290 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 15:00:55 +0200
Subject: [PATCH 07/24] update docks

---
 docs/sources/reference/components/loki/loki.write.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/sources/reference/components/loki/loki.write.md b/docs/sources/reference/components/loki/loki.write.md
index cb1e9090ff0..a9a65ad0091 100644
--- a/docs/sources/reference/components/loki/loki.write.md
+++ b/docs/sources/reference/components/loki/loki.write.md
@@ -171,7 +171,7 @@ The following arguments are supported:
 
 | Name                 | Type       | Description                                                                                                    | Default   | Required |
 | -------------------- | ---------- | -------------------------------------------------------------------------------------------------------------- | --------- | -------- |
-| `drain_timeout`      | `duration` | Maximum time the WAL drain procedure can take, before being forcefully stopped.                                | `"30s"`   | no       |
+| `drain_timeout`      | `duration` | Maximum time the WAL drain procedure can take, before being forcefully stopped.                                | `"15s"`   | no       |
 | `enabled`            | `bool`     | Whether to enable the WAL.                                                                                     | `false`   | no       |
 | `max_read_frequency` | `duration` | Maximum backoff time in the backup read mechanism.                                                             | `"1s"`    | no       |
 | `max_segment_age`    | `duration` | Maximum time a WAL segment should be allowed to live. Segments older than this setting are eventually deleted. | `"1h"`    | no       |

From cb60facd3e66dfeb2a179493ca5b1ab271484daf Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 15:01:03 +0200
Subject: [PATCH 08/24] lint

---
 internal/component/common/loki/client/util_test.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/internal/component/common/loki/client/util_test.go b/internal/component/common/loki/client/util_test.go
index 4a6333abaff..9ad52378db2 100644
--- a/internal/component/common/loki/client/util_test.go
+++ b/internal/component/common/loki/client/util_test.go
@@ -48,5 +48,4 @@ func feedUntilBlocked(t *testing.T, blocked *atomic.Bool, c chan<- loki.Entry) {
 		}
 	}
 	require.True(t, blocked.Load())
-
 }

From bdad34a695f12ab362b2fd63ff8553bb2d1bb179 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 15:01:27 +0200
Subject: [PATCH 09/24] remove unused const

---
 internal/component/common/loki/client/metrics.go | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/internal/component/common/loki/client/metrics.go b/internal/component/common/loki/client/metrics.go
index 1470072ab1d..9f0e68d4782 100644
--- a/internal/component/common/loki/client/metrics.go
+++ b/internal/component/common/loki/client/metrics.go
@@ -15,12 +15,11 @@ const (
 	reasonGeneric       = "ingester_error"
 	reasonRateLimited   = "rate_limited"
 	reasonStreamLimited = "stream_limited"
-	reasonLineTooLong   = "line_too_long"
 	reasonQueueIsFull   = "queue_is_full"
 	reasonBatchTooLarge = "batch_too_large"
 )
 
-var reasons = []string{reasonGeneric, reasonRateLimited, reasonStreamLimited, reasonLineTooLong, reasonQueueIsFull}
+var reasons = []string{reasonGeneric, reasonRateLimited, reasonStreamLimited, reasonQueueIsFull, reasonBatchTooLarge}
 
 type metrics struct {
 	sentBytes                    *prometheus.CounterVec

From 558477c2b508241d46a7549028760a62efd4379a Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Fri, 11 Sep 2026 15:25:24 +0200
Subject: [PATCH 10/24] fix tests

---
 .../common/loki/client/endpoint_test.go       | 20 +++++++++----------
 1 file changed, 10 insertions(+), 10 deletions(-)

diff --git a/internal/component/common/loki/client/endpoint_test.go b/internal/component/common/loki/client/endpoint_test.go
index 49c8c8cfa19..b40a54ff268 100644
--- a/internal/component/common/loki/client/endpoint_test.go
+++ b/internal/component/common/loki/client/endpoint_test.go
@@ -63,8 +63,8 @@ func TestEndpoint(t *testing.T) {
                                loki_write_sent_entries_total{host="__HOST__",tenant=""} 3.0
                                # HELP loki_write_dropped_entries_total Number of log entries dropped because failed to be sent to the ingester after all retries.
                                # TYPE loki_write_dropped_entries_total counter
+                               loki_write_dropped_entries_total{host="__HOST__",reason="batch_too_large",tenant=""} 0
                                loki_write_dropped_entries_total{host="__HOST__",reason="ingester_error",tenant=""} 0
-                               loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant=""} 0
 							   loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant=""} 0
                                loki_write_dropped_entries_total{host="__HOST__",reason="rate_limited",tenant=""} 0
                                loki_write_dropped_entries_total{host="__HOST__",reason="stream_limited",tenant=""} 0
@@ -104,8 +104,8 @@ func TestEndpoint(t *testing.T) {
                               loki_write_sent_entries_total{host="__HOST__",tenant=""} 2.0
                               # HELP loki_write_dropped_entries_total Number of log entries dropped because failed to be sent to the ingester after all retries.
                               # TYPE loki_write_dropped_entries_total counter
+                              loki_write_dropped_entries_total{host="__HOST__",reason="batch_too_large",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="ingester_error",tenant=""} 0
-                              loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant=""} 0
 							  loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="rate_limited",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="stream_limited",tenant=""} 0
@@ -136,8 +136,8 @@ func TestEndpoint(t *testing.T) {
 			expectedMetrics: `
                               # HELP loki_write_dropped_entries_total Number of log entries dropped because failed to be sent to the ingester after all retries.
                               # TYPE loki_write_dropped_entries_total counter
+                              loki_write_dropped_entries_total{host="__HOST__",reason="batch_too_large",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="ingester_error",tenant=""} 1
-                              loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant=""} 0
 							  loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="rate_limited",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="stream_limited",tenant=""} 0
@@ -163,8 +163,8 @@ func TestEndpoint(t *testing.T) {
 			expectedMetrics: `
                               # HELP loki_write_dropped_entries_total Number of log entries dropped because failed to be sent to the ingester after all retries.
                               # TYPE loki_write_dropped_entries_total counter
+                              loki_write_dropped_entries_total{host="__HOST__",reason="batch_too_large",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="ingester_error",tenant=""} 1
-                              loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant=""} 0
 							  loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="rate_limited",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="stream_limited",tenant=""} 0
@@ -198,8 +198,8 @@ func TestEndpoint(t *testing.T) {
 			expectedMetrics: `
                               # HELP loki_write_dropped_entries_total Number of log entries dropped because failed to be sent to the ingester after all retries.
                               # TYPE loki_write_dropped_entries_total counter
+                              loki_write_dropped_entries_total{host="__HOST__",reason="batch_too_large",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="ingester_error",tenant=""} 0
-                              loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant=""} 0
 							  loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="rate_limited",tenant=""} 1
                               loki_write_dropped_entries_total{host="__HOST__",reason="stream_limited",tenant=""} 0
@@ -226,8 +226,8 @@ func TestEndpoint(t *testing.T) {
 			expectedMetrics: `
                               # HELP loki_write_dropped_entries_total Number of log entries dropped because failed to be sent to the ingester after all retries.
                               # TYPE loki_write_dropped_entries_total counter
+                              loki_write_dropped_entries_total{host="__HOST__",reason="batch_too_large",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="ingester_error",tenant=""} 0
-                              loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant=""} 0
 							  loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant=""} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="rate_limited",tenant=""} 1
                               loki_write_dropped_entries_total{host="__HOST__",reason="stream_limited",tenant=""} 0
@@ -257,8 +257,8 @@ func TestEndpoint(t *testing.T) {
                               loki_write_sent_entries_total{host="__HOST__",tenant="tenant-default"} 2.0
                               # HELP loki_write_dropped_entries_total Number of log entries dropped because failed to be sent to the ingester after all retries.
                               # TYPE loki_write_dropped_entries_total counter
+                              loki_write_dropped_entries_total{host="__HOST__", reason="batch_too_large", tenant="tenant-default"} 0
                               loki_write_dropped_entries_total{host="__HOST__", reason="ingester_error", tenant="tenant-default"} 0
-                              loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant="tenant-default"} 0
 							  loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant="tenant-default"} 0
                               loki_write_dropped_entries_total{host="__HOST__", reason="rate_limited", tenant="tenant-default"} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="stream_limited",tenant="tenant-default"} 0
@@ -295,12 +295,12 @@ func TestEndpoint(t *testing.T) {
                               loki_write_sent_entries_total{host="__HOST__",tenant="tenant-default"} 1.0
                               # HELP loki_write_dropped_entries_total Number of log entries dropped because failed to be sent to the ingester after all retries.
                               # TYPE loki_write_dropped_entries_total counter
+                              loki_write_dropped_entries_total{host="__HOST__",reason="batch_too_large",tenant="tenant-1"} 0
+                              loki_write_dropped_entries_total{host="__HOST__",reason="batch_too_large",tenant="tenant-2"} 0
+                              loki_write_dropped_entries_total{host="__HOST__",reason="batch_too_large",tenant="tenant-default"} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="ingester_error",tenant="tenant-1"} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="ingester_error",tenant="tenant-2"} 0
                               loki_write_dropped_entries_total{host="__HOST__",reason="ingester_error",tenant="tenant-default"} 0
-                              loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant="tenant-1"} 0
-                              loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant="tenant-2"} 0
-                              loki_write_dropped_entries_total{host="__HOST__",reason="line_too_long",tenant="tenant-default"} 0
 							  loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant="tenant-1"} 0
 							  loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant="tenant-2"} 0
 							  loki_write_dropped_entries_total{host="__HOST__",reason="queue_is_full",tenant="tenant-default"} 0

From 3063daf72d6e4eb34eafc2974e0a724daa6946b5 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 09:30:03 +0200
Subject: [PATCH 11/24] chore: remove sink and cleanup code

---
 internal/component/loki/write/write.go      | 60 +++++++++------------
 internal/component/loki/write/write_test.go |  8 +--
 2 files changed, 28 insertions(+), 40 deletions(-)

diff --git a/internal/component/loki/write/write.go b/internal/component/loki/write/write.go
index 48e1e186be3..1bd005e5d2a 100644
--- a/internal/component/loki/write/write.go
+++ b/internal/component/loki/write/write.go
@@ -80,18 +80,14 @@ var (
 
 // Component implements the loki.write component.
 type Component struct {
-	opts component.Options
-
-	mut      sync.RWMutex
-	args     Arguments
+	opts     component.Options
 	receiver loki.LogsReceiver
 
+	mut            sync.RWMutex
+	externalLabels model.LabelSet
+
 	// remote write consumer
 	consumer client.Consumer
-
-	// sink is the place where log entries received by this component should be written to.
-	// It will in turn write to client.Consumer.
-	sink loki.EntryHandler
 }
 
 // New creates a new loki.write component.
@@ -116,10 +112,8 @@ func New(o component.Options, args Arguments) (*Component, error) {
 // Run implements component.Component.
 func (c *Component) Run(ctx context.Context) error {
 	defer func() {
-		// First we need to stop the sink. Stopping the sink will not stop the wrapped handler.
-		if c.sink != nil {
-			c.sink.Stop()
-		}
+		c.mut.Lock()
+		defer c.mut.Unlock()
 
 		if c.consumer != nil {
 			if d, ok := c.consumer.(client.DrainableConsumer); ok {
@@ -135,15 +129,8 @@ func (c *Component) Run(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			return nil
-		case entry := <-c.receiver.Chan():
-			c.mut.RLock()
-			select {
-			case <-ctx.Done():
-				c.mut.RUnlock()
-				return nil
-			case c.sink.Chan() <- entry:
-			}
-			c.mut.RUnlock()
+		case e := <-c.receiver.Chan():
+			c.consumeEntry(ctx, e)
 		}
 	}
 }
@@ -158,17 +145,14 @@ func (c *Component) Update(args component.Arguments) error {
 
 	c.mut.Lock()
 	defer c.mut.Unlock()
-	c.args = newArgs
-
-	if c.sink != nil {
-		c.sink.Stop()
-	}
 
 	if c.consumer != nil {
 		// only drain on component shutdown
 		c.consumer.Stop()
 	}
 
+	c.externalLabels = util.MapToModelLabelSet(newArgs.ExternalLabels)
+
 	cfgs := newArgs.convertEndpointConfigs()
 
 	uid := alloyseed.Get().UID
@@ -202,19 +186,23 @@ func (c *Component) Update(args component.Arguments) error {
 		return fmt.Errorf("failed to create cliens: %w", err)
 	}
 
-	c.sink = newEntryHandler(c.consumer, util.MapToModelLabelSet(c.args.ExternalLabels))
-
 	return nil
 }
 
-func newEntryHandler(handler loki.EntryHandler, externalLabels model.LabelSet) loki.EntryHandler {
-	return loki.NewEntryMutatorHandler(handler, func(e loki.Entry) loki.Entry {
-		if len(externalLabels) == 0 {
-			return e
-		}
-		e.Labels = externalLabels.Merge(e.Labels)
-		return e
-	})
+func (c *Component) consumeEntry(ctx context.Context, e loki.Entry) {
+	c.mut.RLock()
+	defer c.mut.RUnlock()
+
+	if len(c.externalLabels) > 0 {
+		e.Labels = c.externalLabels.Merge(e.Labels)
+	}
+
+	select {
+	case <-ctx.Done():
+		return
+	case c.consumer.Chan() <- e:
+		return
+	}
 }
 
 func validateConfigStabilityLevel(o component.Options, args Arguments) error {
diff --git a/internal/component/loki/write/write_test.go b/internal/component/loki/write/write_test.go
index 65fe70e0ac7..c4ac6ca9abc 100644
--- a/internal/component/loki/write/write_test.go
+++ b/internal/component/loki/write/write_test.go
@@ -309,14 +309,14 @@ func testMultipleEndpoint(t *testing.T, alterArgs func(arguments *Arguments)) {
 			require.FailNow(t, "failed waiting for logs")
 		case req := <-ch1:
 			require.Len(t, req.Streams, 1)
-			require.Equal(t, req.Streams[0].Labels, wantLabelSet.Clone().Merge(model.LabelSet{"lbl": "foo"}).String())
+			require.Equal(t, wantLabelSet.Clone().Merge(model.LabelSet{"lbl": "foo"}).String(), req.Streams[0].Labels)
 			require.Len(t, req.Streams[0].Entries, 1)
-			require.Equal(t, req.Streams[0].Entries[0].Line, "writing some text")
+			require.Equal(t, "writing some text", req.Streams[0].Entries[0].Line)
 		case req := <-ch2:
 			require.Len(t, req.Streams, 1)
-			require.Equal(t, req.Streams[0].Labels, wantLabelSet.Clone().Merge(model.LabelSet{"lbl": "bar"}).String())
+			require.Equal(t, wantLabelSet.Clone().Merge(model.LabelSet{"lbl": "bar"}).String(), req.Streams[0].Labels)
 			require.Len(t, req.Streams[0].Entries, 1)
-			require.Equal(t, req.Streams[0].Entries[0].Line, "writing some text")
+			require.Equal(t, "writing some text", req.Streams[0].Entries[0].Line)
 		}
 	}
 }

From fd83d427552a32734a82f0b22f82c3e6a20444ac Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 11:43:12 +0200
Subject: [PATCH 12/24] Refactor: loki client implementations no longer exposed
 channels and return errors

---
 .../component/common/loki/client/consumer.go  |  11 +-
 .../common/loki/client/consumer_fanout.go     |  38 ++--
 .../loki/client/consumer_fanout_test.go       |  43 +++--
 .../common/loki/client/consumer_wal.go        |   7 +-
 .../common/loki/client/consumer_wal_test.go   |  40 ++--
 .../component/common/loki/client/endpoint.go  |  21 ++-
 .../component/common/loki/client/shards.go    |  22 ++-
 .../component/common/loki/client/util_test.go |  11 +-
 .../component/common/loki/wal/watcher_test.go |  20 +-
 internal/component/common/loki/wal/writer.go  | 117 ++++++------
 .../component/common/loki/wal/writer_test.go  | 178 +++++++++++-------
 .../component/loki/source/api/api_test.go     |  82 ++++----
 internal/component/loki/write/write.go        |  10 +-
 13 files changed, 328 insertions(+), 272 deletions(-)

diff --git a/internal/component/common/loki/client/consumer.go b/internal/component/common/loki/client/consumer.go
index 4aee54386bb..d7ed213c9a4 100644
--- a/internal/component/common/loki/client/consumer.go
+++ b/internal/component/common/loki/client/consumer.go
@@ -1,11 +1,14 @@
 package client
 
-import "github.com/grafana/alloy/internal/component/common/loki"
+import (
+	"context"
 
-// Consumer is an interface for consuming Loki log entries. It provides a channel
-// to send entries to and a method to stop the consumer.
+	"github.com/grafana/alloy/internal/component/common/loki"
+)
+
+// Consumer is an interface for consuming Loki log entries.
 type Consumer interface {
-	Chan() chan<- loki.Entry
+	ConsumeEntry(ctx context.Context, entry loki.Entry) error
 	Stop()
 }
 
diff --git a/internal/component/common/loki/client/consumer_fanout.go b/internal/component/common/loki/client/consumer_fanout.go
index d62950d5e8b..eb79fffec76 100644
--- a/internal/component/common/loki/client/consumer_fanout.go
+++ b/internal/component/common/loki/client/consumer_fanout.go
@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"sync"
@@ -19,11 +20,8 @@ func NewFanoutConsumer(logger *slog.Logger, reg prometheus.Registerer, cfgs ...C
 
 	c := &FanoutConsumer{
 		endpoints: make([]*endpoint, 0, len(cfgs)),
-		recv:      make(chan loki.Entry),
 	}
 
-	c.ctx, c.cancel = context.WithCancel(context.Background())
-
 	var (
 		metrics        = newMetrics(reg)
 		endpointsCheck = make(map[string]struct{})
@@ -45,7 +43,6 @@ func NewFanoutConsumer(logger *slog.Logger, reg prometheus.Registerer, cfgs ...C
 		c.endpoints = append(c.endpoints, endpoint)
 	}
 
-	c.wg.Go(c.run)
 	return c, nil
 }
 
@@ -53,36 +50,23 @@ var _ Consumer = (*FanoutConsumer)(nil)
 
 type FanoutConsumer struct {
 	endpoints []*endpoint
-
-	wg     sync.WaitGroup
-	once   sync.Once
-	recv   chan loki.Entry
-	ctx    context.Context
-	cancel context.CancelFunc
 }
 
-func (c *FanoutConsumer) run() {
-	for e := range c.recv {
-		for _, endpoint := range c.endpoints {
-			// NOTE: For now it's fine to ignore error because we can't act on it.
-			_ = endpoint.enqueue(c.ctx, e, 0)
+func (c *FanoutConsumer) ConsumeEntry(ctx context.Context, entry loki.Entry) error {
+	for _, e := range c.endpoints {
+		if err := e.enqueue(ctx, entry, 0); err != nil {
+			// We can receive errQueueIsFull if we have configured endpoint with BlockOnOverflow.
+			// We just skip the endpoint and try the next one.
+			if errors.Is(err, errQueueIsFull) {
+				continue
+			}
+			return err
 		}
 	}
-}
-
-func (c *FanoutConsumer) Chan() chan<- loki.Entry {
-	return c.recv
+	return nil
 }
 
 func (c *FanoutConsumer) Stop() {
-	// First stop the receiving channel.
-	c.once.Do(func() {
-		close(c.recv)
-		c.cancel()
-	})
-
-	c.wg.Wait()
-
 	var stopWG sync.WaitGroup
 	// Stop all endpoints.
 	for _, c := range c.endpoints {
diff --git a/internal/component/common/loki/client/consumer_fanout_test.go b/internal/component/common/loki/client/consumer_fanout_test.go
index c79b0f5df54..956b0cab965 100644
--- a/internal/component/common/loki/client/consumer_fanout_test.go
+++ b/internal/component/common/loki/client/consumer_fanout_test.go
@@ -43,13 +43,20 @@ func TestFanoutConsumer(t *testing.T) {
 	}
 	var totalLines = 100
 	for i := range totalLines {
-		consumer.Chan() <- loki.Entry{
-			Labels: testLabels,
-			Entry: push.Entry{
-				Timestamp: time.Now(),
-				Line:      fmt.Sprintf("line%d", i),
-			},
-		}
+		require.NoError(
+			t,
+			consumer.ConsumeEntry(
+				t.Context(),
+				loki.Entry{
+					Labels: testLabels,
+					Entry: push.Entry{
+						Timestamp: time.Now(),
+						Line:      fmt.Sprintf("line%d", i),
+					},
+				},
+			),
+		)
+
 	}
 
 	require.Eventually(t, func() bool {
@@ -104,13 +111,19 @@ func TestFanoutConsumer_MultipleConfigs(t *testing.T) {
 	}
 	var totalLines = 100
 	for i := range totalLines {
-		consumer.Chan() <- loki.Entry{
-			Labels: testLabels,
-			Entry: push.Entry{
-				Timestamp: time.Now(),
-				Line:      fmt.Sprintf("line%d", i),
-			},
-		}
+		require.NoError(
+			t,
+			consumer.ConsumeEntry(
+				t.Context(),
+				loki.Entry{
+					Labels: testLabels,
+					Entry: push.Entry{
+						Timestamp: time.Now(),
+						Line:      fmt.Sprintf("line%d", i),
+					},
+				},
+			),
+		)
 	}
 
 	// times 2 due to endpoints being run
@@ -238,7 +251,7 @@ func TestFanoutConsumer_StopWithFullSendQueue(t *testing.T) {
 	consumer, err := NewFanoutConsumer(logging.NewSlogNop(), prometheus.NewRegistry(), endpointConfig)
 	require.NoError(t, err)
 
-	feedUntilBlocked(t, blocked, consumer.Chan())
+	feedUntilBlocked(t, blocked, consumer)
 
 	done := make(chan struct{})
 	go func() {
diff --git a/internal/component/common/loki/client/consumer_wal.go b/internal/component/common/loki/client/consumer_wal.go
index 4efd534be7f..fddb8db0e33 100644
--- a/internal/component/common/loki/client/consumer_wal.go
+++ b/internal/component/common/loki/client/consumer_wal.go
@@ -80,7 +80,7 @@ func NewWALConsumer(logger *slog.Logger, reg prometheus.Registerer, walCfg wal.C
 		})
 	}
 
-	writer.Start(walCfg.MaxSegmentAge)
+	writer.Start()
 
 	return m, nil
 }
@@ -109,8 +109,9 @@ type WALConsumer struct {
 	pairs  []endpointWatcherPair
 }
 
-func (m *WALConsumer) Chan() chan<- loki.Entry {
-	return m.writer.Chan()
+// ConsumeEntry implements DrainableConsumer.
+func (m *WALConsumer) ConsumeEntry(ctx context.Context, entry loki.Entry) error {
+	return m.writer.WriteEntry(entry)
 }
 
 func (m *WALConsumer) Stop() {
diff --git a/internal/component/common/loki/client/consumer_wal_test.go b/internal/component/common/loki/client/consumer_wal_test.go
index e5d0f473d4c..afac0b90408 100644
--- a/internal/component/common/loki/client/consumer_wal_test.go
+++ b/internal/component/common/loki/client/consumer_wal_test.go
@@ -58,13 +58,19 @@ func TestWALConsumer(t *testing.T) {
 	}
 	var totalLines = 100
 	for i := range totalLines {
-		consumer.Chan() <- loki.Entry{
-			Labels: testLabels,
-			Entry: push.Entry{
-				Timestamp: time.Now(),
-				Line:      fmt.Sprintf("line%d", i),
-			},
-		}
+		require.NoError(
+			t,
+			consumer.ConsumeEntry(
+				t.Context(),
+				loki.Entry{
+					Labels: testLabels,
+					Entry: push.Entry{
+						Timestamp: time.Now(),
+						Line:      fmt.Sprintf("line%d", i),
+					},
+				},
+			),
+		)
 	}
 
 	require.Eventually(t, func() bool {
@@ -125,13 +131,17 @@ func TestWALConsumer_MultipleConfigs(t *testing.T) {
 	}
 	var totalLines = 100
 	for i := range totalLines {
-		consumer.Chan() <- loki.Entry{
-			Labels: testLabels,
-			Entry: push.Entry{
-				Timestamp: time.Now(),
-				Line:      fmt.Sprintf("line%d", i),
-			},
-		}
+		require.NoError(
+			t,
+			consumer.ConsumeEntry(t.Context(),
+				loki.Entry{
+					Labels: testLabels,
+					Entry: push.Entry{
+						Timestamp: time.Now(),
+						Line:      fmt.Sprintf("line%d", i),
+					},
+				}),
+		)
 	}
 
 	// times 2 due to endpoint being run
@@ -582,7 +592,7 @@ func TestWALConsumer_StopWithFullSendQueue(t *testing.T) {
 	consumer, err := NewWALConsumer(logging.NewSlogNop(), prometheus.NewRegistry(), walConfig, endpointConfig)
 	require.NoError(t, err)
 
-	feedUntilBlocked(t, blocked, consumer.Chan())
+	feedUntilBlocked(t, blocked, consumer)
 
 	stopped := make(chan struct{})
 	go func() {
diff --git a/internal/component/common/loki/client/endpoint.go b/internal/component/common/loki/client/endpoint.go
index c37b40d4d7a..ca4c535e7b6 100644
--- a/internal/component/common/loki/client/endpoint.go
+++ b/internal/component/common/loki/client/endpoint.go
@@ -40,11 +40,10 @@ func newEndpoint(metrics *metrics, cfg Config, logger *slog.Logger, markerHandle
 	return c, nil
 }
 
-var errQueueIsFull = errors.New("queue is full")
-
-// enqueue tries to enqueue an entry. It returns an error if the entry could not be enqueued.
-// errQueueIsFull when the queue is full and BlockOnOverflow is false, or context.Canceled if
-// caller canceled ctx.
+// enqueue tries to enqueue an entry, waiting for room until ctx is done.
+// It returns loki.ErrConsumerStopped when the endpoint is shutting down and waiting cannot help,
+// errQueueIsFull when the queue is full and BlockOnOverflow is not set
+// and the context error when ctx is done before the entry is enqueued.
 func (e *endpoint) enqueue(ctx context.Context, entry loki.Entry, segmentNum int) error {
 	bo := backoff.New(ctx, backoff.Config{
 		MinBackoff: 5 * time.Millisecond,
@@ -54,14 +53,20 @@ func (e *endpoint) enqueue(ctx context.Context, entry loki.Entry, segmentNum int
 	tenantID := getTenantID(e.cfg, entry)
 
 	for bo.Ongoing() {
-		if e.shards.enqueue(tenantID, entry, segmentNum) {
+		err := e.shards.enqueue(tenantID, entry, segmentNum)
+
+		if err == nil {
 			return nil
 		}
 
-		if !e.cfg.QueueConfig.BlockOnOverflow {
+		if errors.Is(err, loki.ErrConsumerStopped) {
+			return err
+		}
+
+		if errors.Is(err, errQueueIsFull) && !e.cfg.QueueConfig.BlockOnOverflow {
 			e.metrics.droppedEntries.WithLabelValues(e.cfg.URL.Host, tenantID, reasonQueueIsFull).Inc()
 			e.metrics.droppedBytes.WithLabelValues(e.cfg.URL.Host, tenantID, reasonQueueIsFull).Add(float64(entry.Size()))
-			return errQueueIsFull
+			return err
 		}
 
 		bo.Wait()
diff --git a/internal/component/common/loki/client/shards.go b/internal/component/common/loki/client/shards.go
index 85321af8f00..96dda2aa546 100644
--- a/internal/component/common/loki/client/shards.go
+++ b/internal/component/common/loki/client/shards.go
@@ -361,13 +361,21 @@ func (s *shards) runShard(q *queue) {
 	}
 }
 
+var errQueueIsFull = errors.New("queue is full")
+
 // enqueue routes a log entry to the appropriate shard based on its label fingerprint.
-// Returns false if we could not enqueue the entry, either because the shard is shutting down or the queue is full.
-// It is up to the caller to retry or drop the entry.
-func (s *shards) enqueue(tenantID string, entry loki.Entry, segmentNum int) bool {
+// Returns loki.ErrConsumerStopped if the shard is shutting down, in which case retrying will never
+// succeed, and errQueueIsFull if the queue is full, which the caller may retry or drop the entry.
+func (s *shards) enqueue(tenantID string, entry loki.Entry, segmentNum int) error {
 	s.mut.Lock()
 	defer s.mut.Unlock()
 
+	select {
+	case <-s.softShutdown:
+		return loki.ErrConsumerStopped
+	default:
+	}
+
 	if _, ok := s.tenants[tenantID]; !ok {
 		s.tenants[tenantID] = struct{}{}
 		s.initBatchMetrics(tenantID)
@@ -376,12 +384,10 @@ func (s *shards) enqueue(tenantID string, entry loki.Entry, segmentNum int) bool
 	fingerprint := entry.Labels.FastFingerprint()
 	shard := uint64(fingerprint) % uint64(len(s.queues))
 
-	select {
-	case <-s.softShutdown:
-		return false
-	default:
-		return s.queues[shard].append(tenantID, entry, segmentNum)
+	if !s.queues[shard].append(tenantID, entry, segmentNum) {
+		return errQueueIsFull
 	}
+	return nil
 }
 
 func (s *shards) initBatchMetrics(tenantID string) {
diff --git a/internal/component/common/loki/client/util_test.go b/internal/component/common/loki/client/util_test.go
index 9ad52378db2..58d7453f3f4 100644
--- a/internal/component/common/loki/client/util_test.go
+++ b/internal/component/common/loki/client/util_test.go
@@ -1,6 +1,7 @@
 package client
 
 import (
+	"context"
 	"net/http"
 	"net/http/httptest"
 	"sync"
@@ -35,17 +36,17 @@ func newBlockedServer() (*httptest.Server, *atomic.Bool, func()) {
 	return server, blocked, release
 }
 
-func feedUntilBlocked(t *testing.T, blocked *atomic.Bool, c chan<- loki.Entry) {
+func feedUntilBlocked(t *testing.T, blocked *atomic.Bool, consumer Consumer) {
 	e := loki.NewEntry(model.LabelSet{"A": "b"}, push.Entry{
 		Line:      "test",
 		Timestamp: time.Now(),
 	})
 
 	for !blocked.Load() {
-		select {
-		case c <- e:
-		case <-time.After(50 * time.Millisecond):
-		}
+		ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
+
+		_ = consumer.ConsumeEntry(ctx, e)
+		cancel()
 	}
 	require.True(t, blocked.Load())
 }
diff --git a/internal/component/common/loki/wal/watcher_test.go b/internal/component/common/loki/wal/watcher_test.go
index 414c1c2f4cd..ae82ee2b05a 100644
--- a/internal/component/common/loki/wal/watcher_test.go
+++ b/internal/component/common/loki/wal/watcher_test.go
@@ -357,7 +357,7 @@ func TestWatcher(t *testing.T) {
 				t,
 				&watcherTestResources{
 					writeEntry: func(entry loki.Entry) {
-						_ = ew.WriteEntry(entry, wl)
+						_ = ew.writeEntry(entry, wl)
 					},
 					notifyWrite: func() {
 						watcher.NotifyWrite()
@@ -433,7 +433,7 @@ func TestWatcher_Replay(t *testing.T) {
 		ew := newEntryWriter()
 
 		// First, write to segment 0. This will be the last "marked" segment
-		err = ew.WriteEntry(loki.Entry{
+		err = ew.writeEntry(loki.Entry{
 			Labels: labels,
 			Entry: push.Entry{
 				Timestamp: time.Now(),
@@ -448,7 +448,7 @@ func TestWatcher_Replay(t *testing.T) {
 
 		// Now, write to segment 1, this will be a segment not marked, hence replayed
 		for _, line := range segment1Lines {
-			err = ew.WriteEntry(loki.Entry{
+			err = ew.writeEntry(loki.Entry{
 				Labels: labels,
 				Entry: push.Entry{
 					Timestamp: time.Now(),
@@ -464,7 +464,7 @@ func TestWatcher_Replay(t *testing.T) {
 
 		// Finally, write some data to the last segment, this will be the write head
 		for _, line := range segment2Lines {
-			err = ew.WriteEntry(loki.Entry{
+			err = ew.writeEntry(loki.Entry{
 				Labels: labels,
 				Entry: push.Entry{
 					Timestamp: time.Now(),
@@ -515,7 +515,7 @@ func TestWatcher_Replay(t *testing.T) {
 		ew := newEntryWriter()
 
 		// First, write to segment 0. This will be the last "marked" segment
-		err = ew.WriteEntry(loki.Entry{
+		err = ew.writeEntry(loki.Entry{
 			Labels: labels,
 			Entry: push.Entry{
 				Timestamp: time.Now(),
@@ -530,7 +530,7 @@ func TestWatcher_Replay(t *testing.T) {
 
 		// Now, write to segment 1, this will be a segment not marked, hence replayed
 		for _, line := range segment1Lines {
-			err = ew.WriteEntry(loki.Entry{
+			err = ew.writeEntry(loki.Entry{
 				Labels: labels,
 				Entry: push.Entry{
 					Timestamp: time.Now(),
@@ -552,7 +552,7 @@ func TestWatcher_Replay(t *testing.T) {
 
 		// Write something after watcher started
 		for _, line := range segment2Lines {
-			err = ew.WriteEntry(loki.Entry{
+			err = ew.writeEntry(loki.Entry{
 				Labels: labels,
 				Entry: push.Entry{
 					Timestamp: time.Now(),
@@ -642,7 +642,7 @@ func TestWatcher_StopAndDrainWAL(t *testing.T) {
 		writeNLines := func(t *testing.T, n int) {
 			for range n {
 				// First, write to segment 0. This will be the last "marked" segment
-				err := ew.WriteEntry(loki.Entry{
+				err := ew.writeEntry(loki.Entry{
 					Labels: labels,
 					Entry: push.Entry{
 						Timestamp: time.Now(),
@@ -694,7 +694,7 @@ func TestWatcher_StopAndDrainWAL(t *testing.T) {
 		writeNLines := func(t *testing.T, n int) {
 			for range n {
 				// First, write to segment 0. This will be the last "marked" segment
-				err := ew.WriteEntry(loki.Entry{
+				err := ew.writeEntry(loki.Entry{
 					Labels: labels,
 					Entry: push.Entry{
 						Timestamp: time.Now(),
@@ -747,7 +747,7 @@ func TestWatcher_StopAndDrainWAL(t *testing.T) {
 		writeNLines := func(t *testing.T, n int) {
 			for range n {
 				// First, write to segment 0. This will be the last "marked" segment
-				err := ew.WriteEntry(loki.Entry{
+				err := ew.writeEntry(loki.Entry{
 					Labels: labels,
 					Entry: push.Entry{
 						Timestamp: time.Now(),
diff --git a/internal/component/common/loki/wal/writer.go b/internal/component/common/loki/wal/writer.go
index f6de1262730..655a4b85c04 100644
--- a/internal/component/common/loki/wal/writer.go
+++ b/internal/component/common/loki/wal/writer.go
@@ -37,16 +37,16 @@ type WriteEventSubscriber interface {
 	NotifyWrite()
 }
 
-// Writer implements loki.EntryHandler, exposing a channel were scraping targets can write to. Reading from there, it
-// writes incoming entries to a WAL.
-// Also, since Writer is responsible for all changing operations over the WAL, therefore a routine is run for cleaning
-// old segments.
+// Writer writes log entries to a write-ahead log. It is also responsible for the WAL's segments,
+// removing those older than the configured maximum age and notifying subscribers when it does.
 type Writer struct {
-	entries     chan loki.Entry
-	logger      *slog.Logger
-	wg          sync.WaitGroup
-	once        sync.Once
-	wal         WAL
+	logger *slog.Logger
+	cfg    Config
+	wg     sync.WaitGroup
+	wal    WAL
+
+	writeMut    sync.Mutex
+	stopped     bool
 	entryWriter *entryWriter
 
 	cleanupSubscribersLock sync.RWMutex
@@ -59,7 +59,7 @@ type Writer struct {
 	lastReclaimedSegment             *prometheus.GaugeVec
 	lastWrittenTimestamp             *prometheus.GaugeVec
 
-	closeCleaner chan struct{}
+	done chan struct{}
 }
 
 // NewWriter creates a new Writer.
@@ -74,12 +74,12 @@ func NewWriter(walCfg Config, logger *slog.Logger, reg prometheus.Registerer) (*
 	}
 
 	wrt := &Writer{
-		entries:      make(chan loki.Entry),
-		logger:       logger,
-		wg:           sync.WaitGroup{},
-		wal:          wl,
-		entryWriter:  newEntryWriter(),
-		closeCleaner: make(chan struct{}, 1),
+		logger:      logger,
+		entryWriter: newEntryWriter(),
+		wg:          sync.WaitGroup{},
+		cfg:         walCfg,
+		wal:         wl,
+		done:        make(chan struct{}, 1),
 	}
 
 	wrt.reclaimedOldSegmentsSpaceCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
@@ -111,64 +111,65 @@ func NewWriter(walCfg Config, logger *slog.Logger, reg prometheus.Registerer) (*
 	return wrt, nil
 }
 
-func (wrt *Writer) Start(maxSegmentAge time.Duration) {
-	// main WAL writer routine
-	wrt.wg.Go(func() {
-		for e := range wrt.entries {
-			if err := wrt.entryWriter.WriteEntry(e, wrt.wal); err != nil {
-				wrt.logger.Error("failed to write entry", "err", err)
-				// if an error occurred while writing the wal, go to next entry and don't notify write subscribers
-				continue
-			}
-
-			// emit metric with latest written timestamp, to be able to track delay from writer to watcher
-			wrt.lastWrittenTimestamp.WithLabelValues().Set(float64(e.Timestamp.Unix()))
-
-			wrt.writeSubscribersLock.RLock()
-			for _, s := range wrt.writeSubscribers {
-				s.NotifyWrite()
-			}
-			wrt.writeSubscribersLock.RUnlock()
-		}
-	})
-
+func (wrt *Writer) Start() {
 	// WAL cleanup routine that cleans old segments
 	wrt.wg.Go(func() {
 		// By cleaning every 10th of the configured threshold for considering a segment old, we are allowing a maximum slip
 		// of 10%. If the configured time is 1 hour, that'd be 6 minutes.
-		triggerEvery := maxSegmentAge / 10
+		triggerEvery := wrt.cfg.MaxSegmentAge / 10
 		if triggerEvery < minimumCleanSegmentsEvery {
 			triggerEvery = minimumCleanSegmentsEvery
 		}
 		trigger := time.NewTicker(triggerEvery)
+		defer trigger.Stop()
 		for {
 			select {
 			case <-trigger.C:
 				wrt.logger.Debug("Running wal old segments cleanup")
-				if err := wrt.cleanSegments(maxSegmentAge); err != nil {
+				if err := wrt.cleanSegments(wrt.cfg.MaxSegmentAge); err != nil {
 					wrt.logger.Error("Error cleaning old segments", "err", err)
 				}
-			case <-wrt.closeCleaner:
-				trigger.Stop()
+			case <-wrt.done:
 				return
 			}
 		}
 	})
 }
 
-func (wrt *Writer) Chan() chan<- loki.Entry {
-	return wrt.entries
+func (wrt *Writer) WriteEntry(entry loki.Entry) error {
+	wrt.writeMut.Lock()
+	defer wrt.writeMut.Unlock()
+
+	if wrt.stopped {
+		return loki.ErrConsumerStopped
+	}
+
+	if err := wrt.entryWriter.writeEntry(entry, wrt.wal); err != nil {
+		return err
+	}
+
+	// emit metric with latest written timestamp, to be able to track delay from writer to watcher
+	wrt.lastWrittenTimestamp.WithLabelValues().Set(float64(entry.Timestamp.Unix()))
+
+	wrt.writeSubscribersLock.RLock()
+	for _, s := range wrt.writeSubscribers {
+		s.NotifyWrite()
+	}
+	wrt.writeSubscribersLock.RUnlock()
+
+	return nil
 }
 
 func (wrt *Writer) Stop() {
-	wrt.once.Do(func() {
-		close(wrt.entries)
-	})
-	// close cleaner routine
-	wrt.closeCleaner <- struct{}{}
-	// Wait for routine to write to wal all pending entries
+	wrt.writeMut.Lock()
+	defer wrt.writeMut.Unlock()
+	wrt.stopped = true
+
+	// Stop cleaner routine and wait for it to stop.
+	close(wrt.done)
 	wrt.wg.Wait()
-	// Close WAL to finalize all pending writes
+
+	// Close WAL to finalize all pending writes.
 	wrt.wal.Close()
 }
 
@@ -251,17 +252,19 @@ func newEntryWriter() *entryWriter {
 	}
 }
 
-// WriteEntry writes a loki.Entry to a WAL. Note that since it's re-using the same Record object for every
-// write, it first has to be reset, and then overwritten accordingly. Therefore, WriteEntry is not thread-safe.
-func (ew *entryWriter) WriteEntry(entry loki.Entry, wl WAL) error {
-	// Reset wal record slices
-	ew.reusableWALRecord.Reset()
+func (ew *entryWriter) writeEntry(entry loki.Entry, wl WAL) error {
+	defer ew.reusableWALRecord.Reset()
 
 	var fp uint64
 	lbs := labels.FromMap(util.ModelLabelSetToMap(entry.Labels))
 	fp, _ = lbs.HashWithoutLabels(nil, []string(nil)...)
 	ref := chunks.HeadSeriesRef(fp)
 
+	ew.reusableWALRecord.Series = append(ew.reusableWALRecord.Series, record.RefSeries{
+		Ref:    ref,
+		Labels: lbs,
+	})
+
 	// Append the entry to an already existing stream (if any)
 	ew.reusableWALRecord.RefEntries = append(ew.reusableWALRecord.RefEntries, RefEntries{
 		Ref: ref,
@@ -270,10 +273,6 @@ func (ew *entryWriter) WriteEntry(entry loki.Entry, wl WAL) error {
 		},
 		Created: entry.Created(),
 	})
-	ew.reusableWALRecord.Series = append(ew.reusableWALRecord.Series, record.RefSeries{
-		Ref:    ref,
-		Labels: lbs,
-	})
 
 	return wl.Log(ew.reusableWALRecord)
 }
diff --git a/internal/component/common/loki/wal/writer_test.go b/internal/component/common/loki/wal/writer_test.go
index 534ca2a2b75..b909861af0c 100644
--- a/internal/component/common/loki/wal/writer_test.go
+++ b/internal/component/common/loki/wal/writer_test.go
@@ -19,47 +19,81 @@ import (
 	autil "github.com/grafana/alloy/internal/util"
 )
 
-func TestWriter_EntriesAreWrittenToWAL(t *testing.T) {
-	dir := t.TempDir()
+func TestWriter(t *testing.T) {
+	t.Run("entries are written to wal", func(t *testing.T) {
+		dir := t.TempDir()
+
+		writer, err := NewWriter(Config{
+			Dir:           dir,
+			Enabled:       true,
+			MaxSegmentAge: time.Minute,
+		}, autil.TestAlloyLogger(t).Slog(), prometheus.NewRegistry())
+		require.NoError(t, err)
+		defer writer.Stop()
+
+		// write entries to wal and sync
+		writer.Start()
+
+		var testLabels = model.LabelSet{
+			"testing": "log",
+		}
+		var lines = []string{
+			"some line",
+			"some other line",
+			"some other other line",
+		}
 
-	writer, err := NewWriter(Config{
-		Dir:           dir,
-		Enabled:       true,
-		MaxSegmentAge: time.Minute,
-	}, autil.TestAlloyLogger(t).Slog(), prometheus.NewRegistry())
-	require.NoError(t, err)
-	defer func() {
-		writer.Stop()
-	}()
-	// write entries to wal and sync
-	writer.Start(time.Minute)
+		for _, line := range lines {
+			require.NoError(t, writer.WriteEntry(
+				loki.Entry{
+					Labels: testLabels,
+					Entry: push.Entry{
+						Timestamp: time.Now(),
+						Line:      line,
+					},
+				},
+			))
+		}
 
-	var testLabels = model.LabelSet{
-		"testing": "log",
-	}
-	var lines = []string{
-		"some line",
-		"some other line",
-		"some other other line",
-	}
+		// accessing the WAL inside, just for testing!
+		require.NoError(t, writer.wal.Sync(), "failed to sync wal")
 
-	for _, line := range lines {
-		writer.Chan() <- loki.Entry{
-			Labels: testLabels,
+		// assert over WAL entries
+		readEntries := eventuallyReadWAL(t, len(lines), dir)
+		require.NotNil(t, readEntries)
+		require.Equal(t, testLabels, readEntries[0].Labels)
+	})
+
+	t.Run("stopped writer returns error", func(t *testing.T) {
+		dir := t.TempDir()
+
+		writer, err := NewWriter(Config{
+			Dir:           dir,
+			Enabled:       true,
+			MaxSegmentAge: time.Minute,
+		}, logging.NewSlogNop(), prometheus.NewRegistry())
+
+		require.NoError(t, err)
+		writer.Start()
+
+		require.NoError(t, writer.WriteEntry(loki.Entry{
+			Labels: model.LabelSet{"key": "value"},
 			Entry: push.Entry{
 				Timestamp: time.Now(),
-				Line:      line,
+				Line:      "lin",
 			},
-		}
-	}
+		}))
 
-	// accessing the WAL inside, just for testing!
-	require.NoError(t, writer.wal.Sync(), "failed to sync wal")
+		writer.Stop()
 
-	// assert over WAL entries
-	readEntries := eventuallyReadWAL(t, len(lines), dir)
-	require.NotNil(t, readEntries)
-	require.Equal(t, testLabels, readEntries[0].Labels)
+		require.ErrorIs(t, writer.WriteEntry(loki.Entry{
+			Labels: model.LabelSet{"key": "value"},
+			Entry: push.Entry{
+				Timestamp: time.Now(),
+				Line:      "lin",
+			},
+		}), loki.ErrConsumerStopped)
+	})
 }
 
 type notifySegmentsCleanedFunc func(num int)
@@ -72,12 +106,12 @@ func (n notifySegmentsCleanedFunc) SeriesReset(segmentNum int) {
 }
 
 func TestWriter_OldSegmentsAreCleanedUp(t *testing.T) {
-	dir := t.TempDir()
-
-	maxSegmentAge := time.Second * 2
-
-	subscriber1 := []int{}
-	subscriber2 := []int{}
+	var (
+		dir           = t.TempDir()
+		subscriber1   = []int{}
+		subscriber2   = []int{}
+		maxSegmentAge = time.Second * 2
+	)
 
 	writer, err := NewWriter(Config{
 		Dir:           dir,
@@ -85,10 +119,8 @@ func TestWriter_OldSegmentsAreCleanedUp(t *testing.T) {
 		MaxSegmentAge: maxSegmentAge,
 	}, autil.TestAlloyLogger(t).Slog(), prometheus.NewRegistry())
 	require.NoError(t, err)
-	writer.Start(maxSegmentAge)
-	defer func() {
-		writer.Stop()
-	}()
+	defer writer.Stop()
+	writer.Start()
 
 	notificationMutex := sync.Mutex{}
 	// add writer events subscriber. Add multiple to test fanout
@@ -114,13 +146,15 @@ func TestWriter_OldSegmentsAreCleanedUp(t *testing.T) {
 	}
 
 	for _, line := range lines {
-		writer.Chan() <- loki.Entry{
-			Labels: testLabels,
-			Entry: push.Entry{
-				Timestamp: time.Now(),
-				Line:      line,
+		require.NoError(t, writer.WriteEntry(
+			loki.Entry{
+				Labels: testLabels,
+				Entry: push.Entry{
+					Timestamp: time.Now(),
+					Line:      line,
+				},
 			},
-		}
+		))
 	}
 
 	// accessing the WAL inside, just for testing!
@@ -166,11 +200,11 @@ func TestWriter_OldSegmentsAreCleanedUp(t *testing.T) {
 }
 
 func TestWriter_NoSegmentIsCleanedUpIfTheresOnlyOne(t *testing.T) {
-	dir := t.TempDir()
-
-	maxSegmentAge := time.Second * 2
-
-	segmentsReclaimedNotificationsReceived := []int{}
+	var (
+		dir                                    = t.TempDir()
+		maxSegmentAge                          = 2 * time.Second
+		segmentsReclaimedNotificationsReceived = []int{}
+	)
 
 	writer, err := NewWriter(Config{
 		Dir:           dir,
@@ -178,7 +212,7 @@ func TestWriter_NoSegmentIsCleanedUpIfTheresOnlyOne(t *testing.T) {
 		MaxSegmentAge: maxSegmentAge,
 	}, autil.TestAlloyLogger(t).Slog(), prometheus.NewRegistry())
 	require.NoError(t, err)
-	writer.Start(maxSegmentAge)
+	writer.Start()
 	defer func() {
 		writer.Stop()
 	}()
@@ -197,13 +231,15 @@ func TestWriter_NoSegmentIsCleanedUpIfTheresOnlyOne(t *testing.T) {
 	}
 
 	for _, line := range lines {
-		writer.Chan() <- loki.Entry{
-			Labels: testLabels,
-			Entry: push.Entry{
-				Timestamp: time.Now(),
-				Line:      line,
+		require.NoError(t, writer.WriteEntry(
+			loki.Entry{
+				Labels: testLabels,
+				Entry: push.Entry{
+					Timestamp: time.Now(),
+					Line:      line,
+				},
 			},
-		}
+		))
 	}
 
 	// accessing the WAL inside, just for testing!
@@ -350,20 +386,22 @@ func benchWriteEntries(b *testing.B, lines, labelSetCount int) {
 		MaxSegmentAge: time.Minute,
 	}, logging.NewSlogNop(), prometheus.NewRegistry())
 	require.NoError(b, err)
-	writer.Start(time.Minute)
+	writer.Start()
 	defer func() {
 		writer.Stop()
 	}()
 
 	for i := 0; i < lines; i++ {
-		writer.Chan() <- loki.Entry{
-			Labels: model.LabelSet{
-				"someLabel": model.LabelValue(fmt.Sprint(i % labelSetCount)),
-			},
-			Entry: push.Entry{
-				Timestamp: time.Now(),
-				Line:      fmt.Sprintf("some line being written %d", i),
+		require.NoError(b, writer.WriteEntry(
+			loki.Entry{
+				Labels: model.LabelSet{
+					"someLabel": model.LabelValue(fmt.Sprint(i % labelSetCount)),
+				},
+				Entry: push.Entry{
+					Timestamp: time.Now(),
+					Line:      fmt.Sprintf("some line being written %d", i),
+				},
 			},
-		}
+		))
 	}
 }
diff --git a/internal/component/loki/source/api/api_test.go b/internal/component/loki/source/api/api_test.go
index f0f305f6e45..b2822ac5c43 100644
--- a/internal/component/loki/source/api/api_test.go
+++ b/internal/component/loki/source/api/api_test.go
@@ -110,14 +110,13 @@ func TestLokiSourceAPI_Simple(t *testing.T) {
 	defer lokiClient.Stop()
 
 	now := time.Now()
-	select {
-	case lokiClient.Chan() <- loki.Entry{
-		Labels: map[model.LabelName]model.LabelValue{"source": "test"},
-		Entry:  push.Entry{Timestamp: now, Line: "hello world!"},
-	}:
-	case <-ctx.Done():
-		t.Fatalf("timed out while sending test entries via loki client")
-	}
+	require.NoError(t, lokiClient.ConsumeEntry(
+		ctx,
+		loki.NewEntry(
+			model.LabelSet{"source": "test"},
+			push.Entry{Timestamp: now, Line: "hello world!"},
+		),
+	))
 
 	require.Eventually(
 		t,
@@ -156,14 +155,13 @@ func TestLokiSourceAPI_Update(t *testing.T) {
 	defer lokiClient.Stop()
 
 	now := time.Now()
-	select {
-	case lokiClient.Chan() <- loki.Entry{
-		Labels: map[model.LabelName]model.LabelValue{"source": "test"},
-		Entry:  push.Entry{Timestamp: now, Line: "hello world!"},
-	}:
-	case <-ctx.Done():
-		t.Fatalf("timed out while sending test entries via loki client")
-	}
+	require.NoError(t, lokiClient.ConsumeEntry(
+		ctx,
+		loki.NewEntry(
+			model.LabelSet{"source": "test"},
+			push.Entry{Timestamp: now, Line: "hello world!"},
+		),
+	))
 
 	require.Eventually(
 		t,
@@ -186,14 +184,14 @@ func TestLokiSourceAPI_Update(t *testing.T) {
 
 	receiver.Clear()
 
-	select {
-	case lokiClient.Chan() <- loki.Entry{
-		Labels: map[model.LabelName]model.LabelValue{"source": "test"},
-		Entry:  push.Entry{Timestamp: now, Line: "hello brave new world!"},
-	}:
-	case <-ctx.Done():
-		t.Fatalf("timed out while sending test entries via loki client")
-	}
+	require.NoError(t, lokiClient.ConsumeEntry(
+		ctx,
+		loki.NewEntry(
+			model.LabelSet{"source": "test"},
+			push.Entry{Timestamp: now, Line: "hello brave new world!"},
+		),
+	))
+
 	require.Eventually(
 		t,
 		func() bool { return len(receiver.Received()) == 1 },
@@ -238,15 +236,14 @@ func TestLokiSourceAPI_FanOut(t *testing.T) {
 
 	const messagesCount = 100
 	for i := range messagesCount {
-		entry := loki.Entry{
-			Labels: map[model.LabelName]model.LabelValue{"source": "test"},
-			Entry:  push.Entry{Line: fmt.Sprintf("test message #%d", i)},
-		}
-		select {
-		case lokiClient.Chan() <- entry:
-		case <-ctx.Done():
-			t.Log("timed out while sending test entries via loki client")
-		}
+
+		require.NoError(t, lokiClient.ConsumeEntry(
+			ctx,
+			loki.NewEntry(
+				model.LabelSet{"source": "test"},
+				push.Entry{Line: fmt.Sprintf("test message #%d", i)},
+			),
+		))
 	}
 
 	require.Eventually(
@@ -380,14 +377,17 @@ func TestLokiSourceAPI_TLS(t *testing.T) {
 	defer lokiClient.Stop()
 
 	now := time.Now()
-	select {
-	case lokiClient.Chan() <- loki.Entry{
-		Labels: map[model.LabelName]model.LabelValue{"source": "test"},
-		Entry:  push.Entry{Timestamp: now, Line: "hello world over TLS!"},
-	}:
-	case <-ctx.Done():
-		t.Fatalf("timed out while sending test entries via TLS loki client")
-	}
+
+	require.NoError(
+		t,
+		lokiClient.ConsumeEntry(
+			ctx,
+			loki.NewEntry(
+				model.LabelSet{"source": "test"},
+				push.Entry{Timestamp: now, Line: "hello world over TLS!"},
+			),
+		),
+	)
 
 	require.Eventually(
 		t,
diff --git a/internal/component/loki/write/write.go b/internal/component/loki/write/write.go
index 1bd005e5d2a..f84cc45f876 100644
--- a/internal/component/loki/write/write.go
+++ b/internal/component/loki/write/write.go
@@ -191,18 +191,14 @@ func (c *Component) Update(args component.Arguments) error {
 
 func (c *Component) consumeEntry(ctx context.Context, e loki.Entry) {
 	c.mut.RLock()
-	defer c.mut.RUnlock()
+	consumer := c.consumer
 
 	if len(c.externalLabels) > 0 {
 		e.Labels = c.externalLabels.Merge(e.Labels)
 	}
+	c.mut.RUnlock()
 
-	select {
-	case <-ctx.Done():
-		return
-	case c.consumer.Chan() <- e:
-		return
-	}
+	_ = consumer.ConsumeEntry(ctx, e)
 }
 
 func validateConfigStabilityLevel(o component.Options, args Arguments) error {

From 6b881c8e6b3eb4ec1d93c43f01d7adfbc21cc493 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 14:12:59 +0200
Subject: [PATCH 13/24] No longer return errQueueIsBlocked

---
 .../component/common/loki/client/consumer_fanout.go |  9 +++------
 .../component/common/loki/client/consumer_wal.go    |  8 ++------
 internal/component/common/loki/client/endpoint.go   | 10 +++++-----
 .../component/common/loki/client/endpoint_test.go   | 13 +++++--------
 4 files changed, 15 insertions(+), 25 deletions(-)

diff --git a/internal/component/common/loki/client/consumer_fanout.go b/internal/component/common/loki/client/consumer_fanout.go
index eb79fffec76..d880cbdedca 100644
--- a/internal/component/common/loki/client/consumer_fanout.go
+++ b/internal/component/common/loki/client/consumer_fanout.go
@@ -2,7 +2,6 @@ package client
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"log/slog"
 	"sync"
@@ -54,12 +53,10 @@ type FanoutConsumer struct {
 
 func (c *FanoutConsumer) ConsumeEntry(ctx context.Context, entry loki.Entry) error {
 	for _, e := range c.endpoints {
+		// NOTE: The only errors we can get are the context error and loki.ErrConsumerStopped.
+		// In both these cases there is no point trying to enqueue for other endpoints.
+
 		if err := e.enqueue(ctx, entry, 0); err != nil {
-			// We can receive errQueueIsFull if we have configured endpoint with BlockOnOverflow.
-			// We just skip the endpoint and try the next one.
-			if errors.Is(err, errQueueIsFull) {
-				continue
-			}
 			return err
 		}
 	}
diff --git a/internal/component/common/loki/client/consumer_wal.go b/internal/component/common/loki/client/consumer_wal.go
index fddb8db0e33..8d75d1cf7a7 100644
--- a/internal/component/common/loki/client/consumer_wal.go
+++ b/internal/component/common/loki/client/consumer_wal.go
@@ -2,7 +2,6 @@ package client
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"log/slog"
 	"sync"
@@ -209,12 +208,9 @@ func (c *walEndpointAdapter) AppendEntries(ctx context.Context, entries wal.RefE
 		for i := range entries.Entries {
 			e := entries.EntryAt(l, i)
 			err := c.endpoint.enqueue(ctx, e, segment)
-			// We can receive errQueueIsFull if we have configured endpoint with BlockOnOverflow.
-			// Here we just skip the entry and try with the next one.
-			if errors.Is(err, errQueueIsFull) {
-				continue
-			}
 
+			// NOTE: The only errors we can get are the context error and loki.ErrConsumerStopped.
+			// In both these cases there is no point trying to enqueue other entries.
 			if err != nil {
 				return err
 			}
diff --git a/internal/component/common/loki/client/endpoint.go b/internal/component/common/loki/client/endpoint.go
index ca4c535e7b6..f5027a6fde3 100644
--- a/internal/component/common/loki/client/endpoint.go
+++ b/internal/component/common/loki/client/endpoint.go
@@ -40,10 +40,10 @@ func newEndpoint(metrics *metrics, cfg Config, logger *slog.Logger, markerHandle
 	return c, nil
 }
 
-// enqueue tries to enqueue an entry, waiting for room until ctx is done.
-// It returns loki.ErrConsumerStopped when the endpoint is shutting down and waiting cannot help,
-// errQueueIsFull when the queue is full and BlockOnOverflow is not set
-// and the context error when ctx is done before the entry is enqueued.
+// enqueue tries to enqueue an entry. It waits for room while BlockOnOverflow
+// is set and drops the entry when it is not. It will return context error
+// if caller cancels context or loki.ErrConsumerStopped if endpoint
+// has been stopped.
 func (e *endpoint) enqueue(ctx context.Context, entry loki.Entry, segmentNum int) error {
 	bo := backoff.New(ctx, backoff.Config{
 		MinBackoff: 5 * time.Millisecond,
@@ -66,7 +66,7 @@ func (e *endpoint) enqueue(ctx context.Context, entry loki.Entry, segmentNum int
 		if errors.Is(err, errQueueIsFull) && !e.cfg.QueueConfig.BlockOnOverflow {
 			e.metrics.droppedEntries.WithLabelValues(e.cfg.URL.Host, tenantID, reasonQueueIsFull).Inc()
 			e.metrics.droppedBytes.WithLabelValues(e.cfg.URL.Host, tenantID, reasonQueueIsFull).Add(float64(entry.Size()))
-			return err
+			return nil
 		}
 
 		bo.Wait()
diff --git a/internal/component/common/loki/client/endpoint_test.go b/internal/component/common/loki/client/endpoint_test.go
index b40a54ff268..64300e25672 100644
--- a/internal/component/common/loki/client/endpoint_test.go
+++ b/internal/component/common/loki/client/endpoint_test.go
@@ -2,7 +2,6 @@ package client
 
 import (
 	"context"
-	"errors"
 	"net/http"
 	"net/http/httptest"
 	"runtime"
@@ -408,13 +407,11 @@ func TestEndpointBlockOnOverflow(t *testing.T) {
 		require.NoError(t, e.enqueue(t.Context(), entry, 0))
 		require.NoError(t, e.enqueue(t.Context(), entry, 0))
 
-		// Which enqueue fails depends on whether the shard worker has already
-		// consumed the queued batch after the third call. If the third call loses that race,
-		// it returns errQueueIsFull, otherwise the fourth call does.
-		err3 := e.enqueue(t.Context(), entry, 0)
-		err4 := e.enqueue(t.Context(), entry, 0)
-		queueIsFull := errors.Is(err3, errQueueIsFull) || errors.Is(err4, errQueueIsFull)
-		require.True(t, queueIsFull, "expected either the third or fourth enqueue to fail with queue full")
+		require.NoError(t, e.enqueue(t.Context(), entry, 0))
+		require.NoError(t, e.enqueue(t.Context(), entry, 0))
+
+		require.Equal(t, 1, testutil.ToFloat64(m.droppedEntries.WithLabelValues(url.Host, "", reasonQueueIsFull)))
+		require.Equal(t, entry.Size(), testutil.ToFloat64(m.droppedBytes.WithLabelValues(url.Host, "", reasonQueueIsFull)))
 	})
 
 	t.Run("should block until queue has space when BlockOnOverflow is true", func(t *testing.T) {

From be67141c2e28e5b141e084686ea7c348f601f3f8 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 14:13:23 +0200
Subject: [PATCH 14/24] Add test for update and shutdown when queue is blocked

---
 internal/component/loki/write/write.go      |   4 +-
 internal/component/loki/write/write_test.go | 108 ++++++++++++++++++++
 2 files changed, 111 insertions(+), 1 deletion(-)

diff --git a/internal/component/loki/write/write.go b/internal/component/loki/write/write.go
index f84cc45f876..79693f9d370 100644
--- a/internal/component/loki/write/write.go
+++ b/internal/component/loki/write/write.go
@@ -183,7 +183,7 @@ func (c *Component) Update(args component.Arguments) error {
 	}
 
 	if err != nil {
-		return fmt.Errorf("failed to create cliens: %w", err)
+		return fmt.Errorf("failed to create clients: %w", err)
 	}
 
 	return nil
@@ -198,6 +198,8 @@ func (c *Component) consumeEntry(ctx context.Context, e loki.Entry) {
 	}
 	c.mut.RUnlock()
 
+	// NOTE: For now it's ok to ignore error here. Error mean the consumer is going away,
+	// either because ctx was canceled or because it has been stopped by a shutdown or an update.
 	_ = consumer.ConsumeEntry(ctx, e)
 }
 
diff --git a/internal/component/loki/write/write_test.go b/internal/component/loki/write/write_test.go
index c4ac6ca9abc..ed1b8a02711 100644
--- a/internal/component/loki/write/write_test.go
+++ b/internal/component/loki/write/write_test.go
@@ -1,6 +1,7 @@
 package write
 
 import (
+	"context"
 	"fmt"
 	"math"
 	"net/http"
@@ -476,3 +477,110 @@ func benchSingleEndpoint(b *testing.B, tc testCase, alterConfig func(arguments *
 		}, time.Minute, time.Second, "haven't seen expected number of lines")
 	}
 }
+
+func TestUpdateWhileSendingIsBlocked(t *testing.T) {
+	for _, walEnabled := range []bool{false, true} {
+		t.Run(fmt.Sprintf("wal_enabled=%t", walEnabled), func(t *testing.T) {
+			ctx, cancel := context.WithCancel(t.Context())
+			defer cancel()
+
+			ctrl, _ := blockedComponent(t, ctx, walEnabled)
+
+			updated := make(chan error, 1)
+			go func() { updated <- ctrl.Update(blockedEndpointArgs(t, "http://localhost:1", walEnabled)) }()
+
+			select {
+			case err := <-updated:
+				require.NoError(t, err)
+			case <-time.After(30 * time.Second):
+				t.Fatal("Update did not return while sending was blocked")
+			}
+		})
+	}
+}
+
+func TestStopWhileSendingIsBlocked(t *testing.T) {
+	for _, walEnabled := range []bool{false, true} {
+		t.Run(fmt.Sprintf("wal_enabled=%t", walEnabled), func(t *testing.T) {
+			ctx, cancel := context.WithCancel(t.Context())
+			_, stopped := blockedComponent(t, ctx, walEnabled)
+			cancel()
+
+			select {
+			case err := <-stopped:
+				require.NoError(t, err)
+			case <-time.After(30 * time.Second):
+				t.Fatal("Run did not return while sending was blocked")
+			}
+		})
+	}
+}
+
+func blockedComponent(t *testing.T, ctx context.Context, walEnabled bool) (*componenttest.Controller, <-chan error) {
+	t.Helper()
+
+	blocked := atomic.NewBool(false)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		blocked.Store(true)
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	t.Cleanup(srv.Close)
+
+	ctrl, err := componenttest.NewControllerFromID(logging.NewSlogNop(), "loki.write")
+	require.NoError(t, err)
+
+	stopped := make(chan error, 1)
+	go func() {
+		stopped <- ctrl.Run(ctx, blockedEndpointArgs(t, srv.URL, walEnabled), func(o component.Options) component.Options {
+			o.MinStability = featuregate.StabilityExperimental
+			return o
+		})
+	}()
+
+	require.NoError(t, ctrl.WaitExports(5*time.Second))
+
+	ch := ctrl.Exports().(Exports).Receiver.Chan()
+
+	go func() {
+		entry := loki.NewEntry(model.LabelSet{"foo": "bar"}, push.Entry{Timestamp: time.Now(), Line: "very important log"})
+		for {
+			select {
+			case ch <- entry:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	require.Eventually(t, blocked.Load, 10*time.Second, 10*time.Millisecond)
+	return ctrl, stopped
+}
+func blockedEndpointArgs(t *testing.T, url string, walEnabled bool) Arguments {
+	t.Helper()
+
+	cfg := fmt.Sprintf(`
+		endpoint {
+			url                 = "%s"
+			batch_size          = "1B"
+			min_backoff_period  = "1ms"
+			max_backoff_period  = "5ms"
+			max_backoff_retries = 0
+
+			queue_config {
+				capacity          = "1B"
+				min_shards        = 1
+				drain_timeout     = "1s"
+				block_on_overflow = true
+			}
+		}
+
+		wal {
+			enabled       = %t
+			drain_timeout = "1s"
+		}
+	`, url, walEnabled)
+
+	var args Arguments
+	require.NoError(t, syntax.Unmarshal([]byte(cfg), &args))
+	return args
+}

From b374571ba2cc2fd432724876477b31aa4b5a5cda Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 14:35:41 +0200
Subject: [PATCH 15/24] return errQueueIsFull

---
 .../common/loki/client/consumer_fanout.go       | 13 ++++++++++---
 .../common/loki/client/consumer_wal.go          | 17 +++++++++++------
 .../component/common/loki/client/endpoint.go    |  8 ++++----
 .../common/loki/client/endpoint_test.go         | 13 ++++++++-----
 4 files changed, 33 insertions(+), 18 deletions(-)

diff --git a/internal/component/common/loki/client/consumer_fanout.go b/internal/component/common/loki/client/consumer_fanout.go
index d880cbdedca..cd94b4046e7 100644
--- a/internal/component/common/loki/client/consumer_fanout.go
+++ b/internal/component/common/loki/client/consumer_fanout.go
@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"sync"
@@ -53,10 +54,16 @@ type FanoutConsumer struct {
 
 func (c *FanoutConsumer) ConsumeEntry(ctx context.Context, entry loki.Entry) error {
 	for _, e := range c.endpoints {
-		// NOTE: The only errors we can get are the context error and loki.ErrConsumerStopped.
-		// In both these cases there is no point trying to enqueue for other endpoints.
+		err := e.enqueue(ctx, entry, 0)
 
-		if err := e.enqueue(ctx, entry, 0); err != nil {
+		if err != nil {
+			// If we get errQueueIsFull we skipped the entry for this endpoints
+			// and should try next.
+			if errors.Is(err, errQueueIsFull) {
+				continue
+			}
+			// The other errors we can get are the context error and loki.ErrConsumerStopped.
+			// In both these cases there is no point trying to enqueue for other endpoints.
 			return err
 		}
 	}
diff --git a/internal/component/common/loki/client/consumer_wal.go b/internal/component/common/loki/client/consumer_wal.go
index 8d75d1cf7a7..7fe6d1d7806 100644
--- a/internal/component/common/loki/client/consumer_wal.go
+++ b/internal/component/common/loki/client/consumer_wal.go
@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"sync"
@@ -206,18 +207,22 @@ func (c *walEndpointAdapter) AppendEntries(ctx context.Context, entries wal.RefE
 
 	if ok {
 		for i := range entries.Entries {
-			e := entries.EntryAt(l, i)
-			err := c.endpoint.enqueue(ctx, e, segment)
+			entry := entries.EntryAt(l, i)
+			err := c.endpoint.enqueue(ctx, entry, segment)
+
+			// If we get errQueueIsFull we skipped the entry and should
+			// not count it as queued and should move on to the next one.
+			if errors.Is(err, errQueueIsFull) {
+				continue
+			}
 
-			// NOTE: The only errors we can get are the context error and loki.ErrConsumerStopped.
-			// In both these cases there is no point trying to enqueue other entries.
 			if err != nil {
 				return err
 			}
 
 			queuedEntries += 1
-			if e.Timestamp.Unix() > maxSeenTimestamp {
-				maxSeenTimestamp = e.Timestamp.Unix()
+			if entry.Timestamp.Unix() > maxSeenTimestamp {
+				maxSeenTimestamp = entry.Timestamp.Unix()
 			}
 		}
 		// update marker with all successfully queued entries.
diff --git a/internal/component/common/loki/client/endpoint.go b/internal/component/common/loki/client/endpoint.go
index f5027a6fde3..520dff8af7b 100644
--- a/internal/component/common/loki/client/endpoint.go
+++ b/internal/component/common/loki/client/endpoint.go
@@ -41,9 +41,9 @@ func newEndpoint(metrics *metrics, cfg Config, logger *slog.Logger, markerHandle
 }
 
 // enqueue tries to enqueue an entry. It waits for room while BlockOnOverflow
-// is set and drops the entry when it is not. It will return context error
-// if caller cancels context or loki.ErrConsumerStopped if endpoint
-// has been stopped.
+// is set and drops the entry when it is not, returning errQueueIsFull. It will
+// return context error if caller cancels context or loki.ErrConsumerStopped
+// if endpoint has been stopped.
 func (e *endpoint) enqueue(ctx context.Context, entry loki.Entry, segmentNum int) error {
 	bo := backoff.New(ctx, backoff.Config{
 		MinBackoff: 5 * time.Millisecond,
@@ -66,7 +66,7 @@ func (e *endpoint) enqueue(ctx context.Context, entry loki.Entry, segmentNum int
 		if errors.Is(err, errQueueIsFull) && !e.cfg.QueueConfig.BlockOnOverflow {
 			e.metrics.droppedEntries.WithLabelValues(e.cfg.URL.Host, tenantID, reasonQueueIsFull).Inc()
 			e.metrics.droppedBytes.WithLabelValues(e.cfg.URL.Host, tenantID, reasonQueueIsFull).Add(float64(entry.Size()))
-			return nil
+			return errQueueIsFull
 		}
 
 		bo.Wait()
diff --git a/internal/component/common/loki/client/endpoint_test.go b/internal/component/common/loki/client/endpoint_test.go
index 64300e25672..b40a54ff268 100644
--- a/internal/component/common/loki/client/endpoint_test.go
+++ b/internal/component/common/loki/client/endpoint_test.go
@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"runtime"
@@ -407,11 +408,13 @@ func TestEndpointBlockOnOverflow(t *testing.T) {
 		require.NoError(t, e.enqueue(t.Context(), entry, 0))
 		require.NoError(t, e.enqueue(t.Context(), entry, 0))
 
-		require.NoError(t, e.enqueue(t.Context(), entry, 0))
-		require.NoError(t, e.enqueue(t.Context(), entry, 0))
-
-		require.Equal(t, 1, testutil.ToFloat64(m.droppedEntries.WithLabelValues(url.Host, "", reasonQueueIsFull)))
-		require.Equal(t, entry.Size(), testutil.ToFloat64(m.droppedBytes.WithLabelValues(url.Host, "", reasonQueueIsFull)))
+		// Which enqueue fails depends on whether the shard worker has already
+		// consumed the queued batch after the third call. If the third call loses that race,
+		// it returns errQueueIsFull, otherwise the fourth call does.
+		err3 := e.enqueue(t.Context(), entry, 0)
+		err4 := e.enqueue(t.Context(), entry, 0)
+		queueIsFull := errors.Is(err3, errQueueIsFull) || errors.Is(err4, errQueueIsFull)
+		require.True(t, queueIsFull, "expected either the third or fourth enqueue to fail with queue full")
 	})
 
 	t.Run("should block until queue has space when BlockOnOverflow is true", func(t *testing.T) {

From af666a5f59ce2dff9d33a95e44e5545c631564c0 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 14:46:40 +0200
Subject: [PATCH 16/24] cleanup

---
 .../common/loki/client/consumer_wal.go        | 60 ++++++++++---------
 1 file changed, 32 insertions(+), 28 deletions(-)

diff --git a/internal/component/common/loki/client/consumer_wal.go b/internal/component/common/loki/client/consumer_wal.go
index 7fe6d1d7806..939db80669a 100644
--- a/internal/component/common/loki/client/consumer_wal.go
+++ b/internal/component/common/loki/client/consumer_wal.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log/slog"
 	"sync"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
@@ -202,39 +203,42 @@ func (c *walEndpointAdapter) AppendEntries(ctx context.Context, entries wal.RefE
 
 	var (
 		queuedEntries    int
-		maxSeenTimestamp int64 = -1
+		maxSeenTimestamp time.Time
 	)
 
-	if ok {
-		for i := range entries.Entries {
-			entry := entries.EntryAt(l, i)
-			err := c.endpoint.enqueue(ctx, entry, segment)
-
-			// If we get errQueueIsFull we skipped the entry and should
-			// not count it as queued and should move on to the next one.
-			if errors.Is(err, errQueueIsFull) {
-				continue
-			}
-
-			if err != nil {
-				return err
-			}
-
-			queuedEntries += 1
-			if entry.Timestamp.Unix() > maxSeenTimestamp {
-				maxSeenTimestamp = entry.Timestamp.Unix()
-			}
-		}
-		// update marker with all successfully queued entries.
-		c.tracker.UpdateReceivedData(segment, queuedEntries)
-	} else {
+	if !ok {
 		// TODO(thepalbi): Add metric here
-		c.logger.Debug("series for entry not found")
+		c.logger.Debug("series for entries not found")
+		return nil
+	}
+
+	for i := range entries.Entries {
+		entry := entries.EntryAt(l, i)
+		err := c.endpoint.enqueue(ctx, entry, segment)
+
+		// If we get errQueueIsFull we skipped the entry and should
+		// not count it as queued and should move on to the next one.
+		if errors.Is(err, errQueueIsFull) {
+			continue
+		}
+
+		if err != nil {
+			return err
+		}
+
+		queuedEntries += 1
+
+		if entry.Timestamp.After(maxSeenTimestamp) {
+			maxSeenTimestamp = entry.Timestamp
+		}
+	}
+	// update tracker with all successfully queued entries.
+	c.tracker.UpdateReceivedData(segment, queuedEntries)
+
+	if queuedEntries > 0 {
+		c.metrics.lastReadTimestamp.WithLabelValues().Set(float64(maxSeenTimestamp.Unix()))
 	}
 
-	// It's safe to assume that upon an AppendEntries call, there will always be at least
-	// one entry.
-	c.metrics.lastReadTimestamp.WithLabelValues().Set(float64(maxSeenTimestamp))
 	return nil
 }
 

From 12b942c370a17573bf3ddfb4f433a2fa56ef130f Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 14:48:06 +0200
Subject: [PATCH 17/24] Add back log

---
 internal/component/common/loki/wal/writer.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/internal/component/common/loki/wal/writer.go b/internal/component/common/loki/wal/writer.go
index 655a4b85c04..a8e9c5bd36a 100644
--- a/internal/component/common/loki/wal/writer.go
+++ b/internal/component/common/loki/wal/writer.go
@@ -145,6 +145,7 @@ func (wrt *Writer) WriteEntry(entry loki.Entry) error {
 	}
 
 	if err := wrt.entryWriter.writeEntry(entry, wrt.wal); err != nil {
+		wrt.logger.Error("failed to write entry", "err", err)
 		return err
 	}
 

From f52ab50a46b046abc2969558df916894a28f89b7 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 15:06:16 +0200
Subject: [PATCH 18/24] lint

---
 internal/component/loki/source/api/api_test.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/internal/component/loki/source/api/api_test.go b/internal/component/loki/source/api/api_test.go
index b2822ac5c43..79d760b289f 100644
--- a/internal/component/loki/source/api/api_test.go
+++ b/internal/component/loki/source/api/api_test.go
@@ -236,7 +236,6 @@ func TestLokiSourceAPI_FanOut(t *testing.T) {
 
 	const messagesCount = 100
 	for i := range messagesCount {
-
 		require.NoError(t, lokiClient.ConsumeEntry(
 			ctx,
 			loki.NewEntry(

From 0591cfd7f0cefe68bf670608619e21b918fc077c Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 15:33:56 +0200
Subject: [PATCH 19/24] lint

---
 internal/component/common/loki/client/consumer_fanout_test.go | 1 -
 1 file changed, 1 deletion(-)

diff --git a/internal/component/common/loki/client/consumer_fanout_test.go b/internal/component/common/loki/client/consumer_fanout_test.go
index 956b0cab965..6cde092616a 100644
--- a/internal/component/common/loki/client/consumer_fanout_test.go
+++ b/internal/component/common/loki/client/consumer_fanout_test.go
@@ -56,7 +56,6 @@ func TestFanoutConsumer(t *testing.T) {
 				},
 			),
 		)
-
 	}
 
 	require.Eventually(t, func() bool {

From 94d7798877e52a96a9cad5e8d052f0b7bde64923 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 17:08:14 +0200
Subject: [PATCH 20/24] fix chan

---
 internal/component/common/loki/wal/writer.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/internal/component/common/loki/wal/writer.go b/internal/component/common/loki/wal/writer.go
index 6e5670b80d8..39d3ab11f87 100644
--- a/internal/component/common/loki/wal/writer.go
+++ b/internal/component/common/loki/wal/writer.go
@@ -77,7 +77,7 @@ func NewWriter(walCfg Config, logger *slog.Logger, reg prometheus.Registerer, me
 		wg:          sync.WaitGroup{},
 		cfg:         walCfg,
 		wal:         wl,
-		done:        make(chan struct{}, 1),
+		done:        make(chan struct{}),
 		metrics:     metrics,
 	}
 

From 61b650c7d1f645134c039667f599f3bba33e3903 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Mon, 14 Sep 2026 17:39:37 +0200
Subject: [PATCH 21/24] trigger ci


From d5842345d55331d2629ed1c76adcb87a743987cb Mon Sep 17 00:00:00 2001
From: Karl Persson <23356117+kalleep@users.noreply.github.com>
Date: Tue, 15 Sep 2026 15:31:07 +0200
Subject: [PATCH 22/24] Update
 internal/component/common/loki/client/consumer_fanout.go

Co-authored-by: Kyle Eckhart <kgeckhart@users.noreply.github.com>
---
 internal/component/common/loki/client/consumer_fanout.go | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/internal/component/common/loki/client/consumer_fanout.go b/internal/component/common/loki/client/consumer_fanout.go
index cd94b4046e7..5805f067b8e 100644
--- a/internal/component/common/loki/client/consumer_fanout.go
+++ b/internal/component/common/loki/client/consumer_fanout.go
@@ -62,8 +62,7 @@ func (c *FanoutConsumer) ConsumeEntry(ctx context.Context, entry loki.Entry) err
 			if errors.Is(err, errQueueIsFull) {
 				continue
 			}
-			// The other errors we can get are the context error and loki.ErrConsumerStopped.
-			// In both these cases there is no point trying to enqueue for other endpoints.
+			// For any other error we assume it's not useful to send to the next endpoint
 			return err
 		}
 	}

From 43ed462837693da97ccde1538ddfaddf44d930f6 Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Tue, 15 Sep 2026 15:39:32 +0200
Subject: [PATCH 23/24] Add test

---
 .../component/common/loki/client/endpoint_test.go | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/internal/component/common/loki/client/endpoint_test.go b/internal/component/common/loki/client/endpoint_test.go
index b40a54ff268..9b039fdd057 100644
--- a/internal/component/common/loki/client/endpoint_test.go
+++ b/internal/component/common/loki/client/endpoint_test.go
@@ -578,6 +578,21 @@ func TestEndpointCallerCancel(t *testing.T) {
 	})
 }
 
+func TestEndpointStopped(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
+	var url flagext.URLValue
+	require.NoError(t, url.Set(srv.URL))
+
+	e, err := newEndpoint(newMetrics(prometheus.NewRegistry()), Config{
+		URL: url,
+	}, logging.NewSlogNop(), marker.NewNopTracker())
+	require.NoError(t, err)
+	e.stop()
+
+	entry := loki.Entry{Entry: push.Entry{Line: "my entry"}}
+	require.ErrorIs(t, e.enqueue(t.Context(), entry, 0), loki.ErrConsumerStopped)
+}
+
 // histogramSumAndCount returns the sum and count of the single series of the
 // named histogram in reg.
 func histogramSumAndCount(t *testing.T, reg *prometheus.Registry, name string) (float64, uint64) {

From 840c1407994680e5a32d04374ed250656b87f48b Mon Sep 17 00:00:00 2001
From: Kalle <23356117+kalleep@users.noreply.github.com>
Date: Tue, 15 Sep 2026 15:45:00 +0200
Subject: [PATCH 24/24] add guard

---
 internal/component/common/loki/client/util_test.go | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/internal/component/common/loki/client/util_test.go b/internal/component/common/loki/client/util_test.go
index 58d7453f3f4..0444d0b7c57 100644
--- a/internal/component/common/loki/client/util_test.go
+++ b/internal/component/common/loki/client/util_test.go
@@ -10,7 +10,6 @@ import (
 
 	"github.com/grafana/loki/pkg/push"
 	"github.com/prometheus/common/model"
-	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
 
 	"github.com/grafana/alloy/internal/component/common/loki"
@@ -37,16 +36,24 @@ func newBlockedServer() (*httptest.Server, *atomic.Bool, func()) {
 }
 
 func feedUntilBlocked(t *testing.T, blocked *atomic.Bool, consumer Consumer) {
+	t.Helper()
+
+	const timeout = 10 * time.Second
+
 	e := loki.NewEntry(model.LabelSet{"A": "b"}, push.Entry{
 		Line:      "test",
 		Timestamp: time.Now(),
 	})
 
+	deadline := time.Now().Add(timeout)
 	for !blocked.Load() {
+		if time.Now().After(deadline) {
+			t.Fatalf("endpoint did not block within %s", timeout)
+		}
+
 		ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
 
 		_ = consumer.ConsumeEntry(ctx, e)
 		cancel()
 	}
-	require.True(t, blocked.Load())
 }