fix(loki.source.docker): Preserve positions for stopped containers

internal/component/loki/source/docker/tailer.go

@@ -177,11 +177,27 @@ func (t *tailer) stop() {
 		t.wg.Wait()
 		t.logger.Debug("stopped Docker target", "container", t.containerID)
 
-		// If the component is not stopping, then it means that the target for this component is gone and that
-		// we should clear the entry from the positions file.
-		if !t.componentStopping() {
-			t.positions.Remove(positions.CursorKey(t.containerID), t.labelsStr)
-		}
+		t.removePositionIfContainerDeleted()
+	}
+}
+
+func (t *tailer) removePositionIfContainerDeleted() {
+	if t.componentStopping() || t.client == nil {
+		return
+	}
+
+	_, err := t.client.ContainerInspect(context.Background(), t.containerID, client.ContainerInspectOptions{})
+	switch {
+	case err == nil:
+		// The target may have disappeared because the container stopped. Keep its
+		// position so a later restart resumes where the tailer left off.
+		return
+	case cerrdefs.IsNotFound(err):
+		t.positions.Remove(positions.CursorKey(t.containerID), t.labelsStr)
+	default:
+		// Preserve the position when the container's state cannot be determined.
+		// Re-reading logs is more harmful than retaining a stale cursor.
+		t.logger.Warn("could not determine whether to remove Docker container position", "id", t.containerID, "error", err)
 	}
 }
 

internal/component/loki/source/docker/tailer_test.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"encoding/binary"
 	"encoding/json"
+	"errors"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -265,6 +266,66 @@ func TestTailerStopsWhenContainerNotFound(t *testing.T) {
 	}
 }
 
+func TestTailerPositionCleanup(t *testing.T) {
+	tests := []struct {
+		name              string
+		componentStopping bool
+		inspectErr        error
+		wantRemoved       bool
+	}{
+		{
+			name: "stopped container preserves position",
+		},
+		{
+			name:        "removed container deletes position",
+			inspectErr:  cerrdefs.ErrNotFound,
+			wantRemoved: true,
+		},
+		{
+			name:       "inspect error preserves position",
+			inspectErr: errors.New("inspect failed"),
+		},
+		{
+			name:              "component shutdown preserves position",
+			componentStopping: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pos := &trackingPositions{Positions: positions.NewNop()}
+			mock := clientMock{
+				running:    func() bool { return false },
+				finishedAt: func() string { return "0001-01-01T00:00:00Z" },
+				inspectErr: func() error { return tt.inspectErr },
+			}
+			tailer := &tailer{
+				logger:            logging.NewSlogNop(),
+				positions:         pos,
+				containerID:       "container-id",
+				labelsStr:         "{}",
+				client:            mock,
+				componentStopping: func() bool { return tt.componentStopping },
+				running:           true,
+				cancel:            func() {},
+			}
+
+			tailer.stop()
+
+			require.Equal(t, tt.wantRemoved, pos.removed)
+		})
+	}
+}
+
+type trackingPositions struct {
+	positions.Positions
+	removed bool
+}
+
+func (p *trackingPositions) Remove(_, _ string) {
+	p.removed = true
+}
+
 var _ io.ReadCloser = (*stringReader)(nil)
 
 func newStringReader(s string) *stringReader {