diff --git a/packages/suite-base/src/components/PanelExtensionAdapter/PanelExtensionAdapter.tsx b/packages/suite-base/src/components/PanelExtensionAdapter/PanelExtensionAdapter.tsx
index 2be03991d7..6e5b1ca4e5 100644
--- a/packages/suite-base/src/components/PanelExtensionAdapter/PanelExtensionAdapter.tsx
+++ b/packages/suite-base/src/components/PanelExtensionAdapter/PanelExtensionAdapter.tsx
@@ -62,10 +62,10 @@ import { assertNever } from "@lichtblick/suite-base/util/assertNever";
 import { maybeCast } from "@lichtblick/suite-base/util/maybeCast";
 
 import { PanelConfigVersionError } from "./PanelConfigVersionError";
-import { createMessageRangeIterator } from "./messageRangeIterator";
 import { RenderStateConfig, initRenderStateBuilder } from "./renderState";
 import { BuiltinPanelExtensionContext, MessageConverterAlertHandler } from "./types";
 import { useSharedPanelState } from "./useSharedPanelState";
+import { useSubscribeMessageRange } from "./useSubscribeMessageRange";
 
 const log = Logger.getLogger(__filename);
 
@@ -134,7 +134,6 @@ function PanelExtensionAdapter(
     getMetadata,
     sortedTopics,
     sortedServices,
-    getBatchIterator,
   } = messagePipelineContext;
 
   const { capabilities, profile: dataSourceProfile, presence: playerPresence } = playerState;
@@ -202,6 +201,8 @@ function PanelExtensionAdapter(
     [setAlert],
   );
 
+  const subscribeMessageRange = useSubscribeMessageRange(emitMessageConverterAlert);
+
   // Register handlers to update the app settings we subscribe to
   useEffect(() => {
     const handlers = new Map<string, (newValue: AppSettingValue) => void>();
@@ -606,32 +607,11 @@ function PanelExtensionAdapter(
        * - Error handling is still being refined
        * - API surface may change based on testing feedback
        */
-      unstable_subscribeMessageRange({ topic, convertTo, onNewRangeIterator }) {
+      unstable_subscribeMessageRange(args) {
         if (!isMounted()) {
           return () => {};
         }
-
-        const rawBatchIterator = getBatchIterator(topic);
-        if (!rawBatchIterator) {
-          // If no batch iterator is available, just return an empty cleanup function
-          return () => {};
-        }
-
-        const { iterable: messageEventIterable, cancel } = createMessageRangeIterator({
-          topic,
-          convertTo,
-          rawBatchIterator,
-          sortedTopics,
-          messageConverters: messageConverters ?? [],
-          emitAlert: emitMessageConverterAlert,
-        });
-
-        // Call the callback with the processed iterable
-        onNewRangeIterator(messageEventIterable).catch((err: unknown) => {
-          log.error("Error in onNewRangeIterator callback:", err);
-        });
-
-        return cancel;
+        return subscribeMessageRange(args);
       },
 
       unstable_setMessagePathDropConfig(dropConfig) {
@@ -660,7 +640,7 @@ function PanelExtensionAdapter(
     updatePanelSettingsTree,
     setDefaultPanelTitle,
     setMessagePathDropConfig,
-    emitMessageConverterAlert,
+    subscribeMessageRange,
   ]);
 
   const panelContainerRef = useRef<HTMLDivElement>(ReactNull);
diff --git a/packages/suite-base/src/components/PanelExtensionAdapter/index.ts b/packages/suite-base/src/components/PanelExtensionAdapter/index.ts
index 29efb6bf8e..389c157755 100644
--- a/packages/suite-base/src/components/PanelExtensionAdapter/index.ts
+++ b/packages/suite-base/src/components/PanelExtensionAdapter/index.ts
@@ -6,6 +6,7 @@
 // file, You can obtain one at http://mozilla.org/MPL/2.0/
 
 export { default as PanelExtensionAdapter } from "./PanelExtensionAdapter";
+export { useSubscribeMessageRange } from "./useSubscribeMessageRange";
 export type {
   Asset,
   BuiltinPanelExtensionContext,
diff --git a/packages/suite-base/src/components/PanelExtensionAdapter/useSubscribeMessageRange.ts b/packages/suite-base/src/components/PanelExtensionAdapter/useSubscribeMessageRange.ts
new file mode 100644
index 0000000000..6b633f2355
--- /dev/null
+++ b/packages/suite-base/src/components/PanelExtensionAdapter/useSubscribeMessageRange.ts
@@ -0,0 +1,71 @@
+// SPDX-FileCopyrightText: Copyright (C) 2023-2026 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)<lichtblick@bmwgroup.com>
+// SPDX-License-Identifier: MPL-2.0
+
+import { useCallback, useRef } from "react";
+
+import Logger from "@lichtblick/log";
+import { SubscribeMessageRangeArgs } from "@lichtblick/suite";
+import { useMessagePipelineGetter } from "@lichtblick/suite-base/components/MessagePipeline";
+import {
+  ExtensionCatalog,
+  useExtensionCatalog,
+} from "@lichtblick/suite-base/context/ExtensionCatalogContext";
+
+import { createMessageRangeIterator } from "./messageRangeIterator";
+import { MessageConverterAlertHandler } from "./types";
+
+const log = Logger.getLogger(__filename);
+
+const selectInstalledMessageConverters = (state: ExtensionCatalog) =>
+  state.installedMessageConverters;
+
+export type UseSubscribeMessageRange = (args: SubscribeMessageRangeArgs) => () => void;
+
+/**
+ * Returns a stable callback that can be used to subscribe to a message range for a topic.
+ * This centralizes the logic of `unstable_subscribeMessageRange` so it can be used both by
+ * PanelExtensionAdapter and directly by built-in panels (e.g. Plot) without requiring migration
+ * to the PanelExtensionContext API.
+ */
+export function useSubscribeMessageRange(
+  emitAlert?: MessageConverterAlertHandler,
+): UseSubscribeMessageRange {
+  // useMessagePipelineGetter returns a stable getter, so it's safe to call it inside the callback without adding it to dependencies.
+  const getMessagePipelineContext = useMessagePipelineGetter();
+
+  // Keep messageConverters in a ref so changing converters don't invalidate the callback.
+  const messageConverters = useExtensionCatalog(selectInstalledMessageConverters);
+  const messageConvertersRef = useRef(messageConverters);
+  messageConvertersRef.current = messageConverters;
+
+  // Similarly keep emitAlert in a ref so the caller can update it without breaking stability.
+  const emitAlertRef = useRef(emitAlert);
+  emitAlertRef.current = emitAlert;
+
+  return useCallback(
+    ({ topic, convertTo, onNewRangeIterator }: SubscribeMessageRangeArgs) => {
+      const { sortedTopics, getBatchIterator } = getMessagePipelineContext();
+
+      const rawBatchIterator = getBatchIterator(topic);
+      if (!rawBatchIterator) {
+        return () => {};
+      }
+
+      const { iterable: messageEventIterable, cancel } = createMessageRangeIterator({
+        topic,
+        convertTo,
+        rawBatchIterator,
+        sortedTopics,
+        messageConverters: messageConvertersRef.current ?? [],
+        emitAlert: emitAlertRef.current,
+      });
+
+      onNewRangeIterator(messageEventIterable).catch((err: unknown) => {
+        log.error("Error in useSubscribeMessageRange onNewRangeIterator:", err);
+      });
+
+      return cancel;
+    },
+    [getMessagePipelineContext], // getMessagePipelineContext is already stable, but listed for clarity
+  );
+}
diff --git a/packages/suite-base/src/components/PanelExtensionAdapter/useSubscriberMessageRange.test.ts b/packages/suite-base/src/components/PanelExtensionAdapter/useSubscriberMessageRange.test.ts
new file mode 100644
index 0000000000..5cf2913f7f
--- /dev/null
+++ b/packages/suite-base/src/components/PanelExtensionAdapter/useSubscriberMessageRange.test.ts
@@ -0,0 +1,114 @@
+/** @jest-environment jsdom */
+
+// SPDX-FileCopyrightText: Copyright (C) 2023-2026 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)<lichtblick@bmwgroup.com>
+// SPDX-License-Identifier: MPL-2.0
+
+import { renderHook, act } from "@testing-library/react";
+
+import { MessageEvent } from "@lichtblick/suite";
+import { useMessagePipelineGetter } from "@lichtblick/suite-base/components/MessagePipeline";
+import { useExtensionCatalog } from "@lichtblick/suite-base/context/ExtensionCatalogContext";
+import { BasicBuilder } from "@lichtblick/test-builders";
+
+import { createMessageRangeIterator } from "./messageRangeIterator";
+import { useSubscribeMessageRange } from "./useSubscribeMessageRange";
+
+jest.mock("@lichtblick/suite-base/components/MessagePipeline", () => ({
+  useMessagePipelineGetter: jest.fn(),
+}));
+
+jest.mock("@lichtblick/suite-base/context/ExtensionCatalogContext", () => ({
+  useExtensionCatalog: jest.fn(),
+}));
+
+jest.mock("./messageRangeIterator", () => ({
+  createMessageRangeIterator: jest.fn(),
+}));
+
+const mockUseMessagePipelineGetter = useMessagePipelineGetter as jest.Mock;
+const mockUseExtensionCatalog = useExtensionCatalog as jest.Mock;
+const mockCreateMessageRangeIterator = createMessageRangeIterator as jest.Mock;
+
+describe("useSubscribeMessageRange", () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    mockUseExtensionCatalog.mockReturnValue([]);
+  });
+
+  it("does not call onNewRangeIterator when batch iterator is unavailable", () => {
+    // Given
+    mockUseMessagePipelineGetter.mockReturnValue(
+      jest.fn().mockReturnValue({
+        sortedTopics: [],
+        getBatchIterator: jest.fn().mockReturnValue(undefined),
+      }),
+    );
+    const onNewRangeIterator = jest.fn().mockResolvedValue(undefined);
+    const { result } = renderHook(() => useSubscribeMessageRange());
+
+    // When
+    act(() => {
+      result.current({ topic: BasicBuilder.string(), onNewRangeIterator });
+    });
+
+    // Then
+    expect(onNewRangeIterator).not.toHaveBeenCalled();
+  });
+
+  it("returns a callable cancel function when batch iterator is unavailable", () => {
+    // Given
+    mockUseMessagePipelineGetter.mockReturnValue(
+      jest.fn().mockReturnValue({
+        sortedTopics: [],
+        getBatchIterator: jest.fn().mockReturnValue(undefined),
+      }),
+    );
+    const onNewRangeIterator = jest.fn().mockResolvedValue(async () => {});
+    const { result } = renderHook(() => useSubscribeMessageRange());
+    let cancel!: () => void;
+
+    // When
+    act(() => {
+      cancel = result.current({ topic: BasicBuilder.string(), onNewRangeIterator });
+    });
+
+    // Then
+    expect(() => {
+      cancel();
+    }).not.toThrow();
+  });
+
+  it("calls onNewRangeIterator with the iterable when batch iterator is available", () => {
+    // Given
+    const topic = BasicBuilder.string();
+    const mockIterable: AsyncIterable<MessageEvent[]> = { [Symbol.asyncIterator]: jest.fn() };
+    const mockCancel = jest.fn();
+    let cancel!: () => void;
+    const mockBatchIterator = { [Symbol.asyncIterator]: jest.fn() };
+    const onNewRangeIterator = jest.fn().mockResolvedValue(async () => {});
+    mockCreateMessageRangeIterator.mockReturnValue({ iterable: mockIterable, cancel: mockCancel });
+    const mockGetBatchIterator = jest.fn().mockReturnValue(mockBatchIterator);
+    mockUseMessagePipelineGetter.mockReturnValue(
+      jest.fn().mockReturnValue({ sortedTopics: [], getBatchIterator: mockGetBatchIterator }),
+    );
+
+    const { result } = renderHook(() => useSubscribeMessageRange());
+
+    // When
+    act(() => {
+      cancel = result.current({ topic, onNewRangeIterator });
+    });
+
+    // Then
+    expect(mockGetBatchIterator).toHaveBeenCalledWith(topic);
+    expect(onNewRangeIterator).toHaveBeenCalledWith(mockIterable);
+    expect(cancel).toBe(mockCancel);
+    expect(mockCreateMessageRangeIterator).toHaveBeenCalledWith(
+      expect.objectContaining({
+        topic,
+        rawBatchIterator: mockBatchIterator,
+        sortedTopics: [],
+      }),
+    );
+  });
+});
diff --git a/packages/suite-base/src/panels/Plot/Plot.test.tsx b/packages/suite-base/src/panels/Plot/Plot.test.tsx
index be892efe10..3c3a5228b5 100644
--- a/packages/suite-base/src/panels/Plot/Plot.test.tsx
+++ b/packages/suite-base/src/panels/Plot/Plot.test.tsx
@@ -25,6 +25,11 @@ jest.mock("@lichtblick/suite-base/components/PanelContext", () => ({
 
 const mockGetMessagePipelineState = jest.fn();
 const mockSubscribeMessagePipeline = jest.fn();
+const mockSubscribeMessageRange = jest.fn();
+jest.mock("@lichtblick/suite-base/components/PanelExtensionAdapter", () => ({
+  useSubscribeMessageRange: () => mockSubscribeMessageRange,
+}));
+
 jest.mock("@lichtblick/suite-base/components/MessagePipeline", () => ({
   useMessagePipelineGetter: () => mockGetMessagePipelineState,
   useMessagePipelineSubscribe: () => mockSubscribeMessagePipeline,
@@ -88,7 +93,9 @@ jest.mock("./hooks/usePlotInteractionHandlers", () => ({
 let mockCoordinatorInstance: any;
 const mockPlotCoordinatorCtor = jest.fn();
 jest.mock("./PlotCoordinator", () => ({
-  PlotCoordinator: jest.fn((renderer, builder) => mockPlotCoordinatorCtor(renderer, builder)),
+  PlotCoordinator: jest.fn((renderer, builder, subscribeMessageRange) =>
+    mockPlotCoordinatorCtor(renderer, builder, subscribeMessageRange),
+  ),
 }));
 
 const rendererStub = { id: "renderer" } as any;
@@ -312,7 +319,11 @@ describe("Plot Component", () => {
     unmount();
 
     // Then
-    expect(mockPlotCoordinatorCtor).toHaveBeenCalledWith(rendererStub, datasetsBuilderStub);
+    expect(mockPlotCoordinatorCtor).toHaveBeenCalledWith(
+      rendererStub,
+      datasetsBuilderStub,
+      mockSubscribeMessageRange,
+    );
     expect(mockCoordinatorInstance.setSize).toHaveBeenCalledWith({
       width: expect.any(Number),
       height: expect.any(Number),
diff --git a/packages/suite-base/src/panels/Plot/Plot.tsx b/packages/suite-base/src/panels/Plot/Plot.tsx
index 701958782a..201fd9655f 100644
--- a/packages/suite-base/src/panels/Plot/Plot.tsx
+++ b/packages/suite-base/src/panels/Plot/Plot.tsx
@@ -19,6 +19,7 @@ import {
 } from "@lichtblick/suite-base/components/MessagePipeline";
 import { usePanelContext } from "@lichtblick/suite-base/components/PanelContext";
 import { PanelContextMenu } from "@lichtblick/suite-base/components/PanelContextMenu";
+import { useSubscribeMessageRange } from "@lichtblick/suite-base/components/PanelExtensionAdapter";
 import PanelToolbar from "@lichtblick/suite-base/components/PanelToolbar";
 import { PANEL_TOOLBAR_MIN_HEIGHT } from "@lichtblick/suite-base/components/PanelToolbar/constants";
 import Stack from "@lichtblick/suite-base/components/Stack";
@@ -69,6 +70,7 @@ const Plot = (props: PlotProps): React.JSX.Element => {
   const { globalVariables } = useGlobalVariables();
   const getMessagePipelineState = useMessagePipelineGetter();
   const subscribeMessagePipeline = useMessagePipelineSubscribe();
+  const subscribeMessageRange = useSubscribeMessageRange();
 
   const {
     onMouseMove,
@@ -79,7 +81,7 @@ const Plot = (props: PlotProps): React.JSX.Element => {
     onClickPath,
     focusedPath,
     keyDownHandlers,
-    keyUphandlers,
+    keyUpHandlers,
     getPanelContextMenuItems,
   } = usePlotInteractionHandlers({
     config,
@@ -164,7 +166,7 @@ const Plot = (props: PlotProps): React.JSX.Element => {
 
     const contentRect = canvasDiv.getBoundingClientRect();
 
-    const plotCoordinator = new PlotCoordinator(renderer, datasetsBuilder);
+    const plotCoordinator = new PlotCoordinator(renderer, datasetsBuilder, subscribeMessageRange);
     setCoordinator(plotCoordinator);
 
     plotCoordinator.setSize({
@@ -188,7 +190,7 @@ const Plot = (props: PlotProps): React.JSX.Element => {
       resizeObserver.disconnect();
       plotCoordinator.destroy();
     };
-  }, [canvasDiv, datasetsBuilder, renderer]);
+  }, [canvasDiv, datasetsBuilder, renderer, subscribeMessageRange]);
 
   const numSeries = config.paths.length;
   const tooltipContent = useMemo(() => {
@@ -287,7 +289,7 @@ const Plot = (props: PlotProps): React.JSX.Element => {
         )}
         <PanelContextMenu getItems={getPanelContextMenuItems} />
       </Stack>
-      <KeyListener global keyDownHandlers={keyDownHandlers} keyUpHandlers={keyUphandlers} />
+      <KeyListener global keyDownHandlers={keyDownHandlers} keyUpHandlers={keyUpHandlers} />
     </Stack>
   );
 };
diff --git a/packages/suite-base/src/panels/Plot/PlotCoordinator.test.ts b/packages/suite-base/src/panels/Plot/PlotCoordinator.test.ts
index 24d756651c..41a9001015 100644
--- a/packages/suite-base/src/panels/Plot/PlotCoordinator.test.ts
+++ b/packages/suite-base/src/panels/Plot/PlotCoordinator.test.ts
@@ -8,9 +8,9 @@ import { parseMessagePath } from "@lichtblick/message-path";
 import { simpleGetMessagePathDataItems } from "@lichtblick/suite-base/components/MessagePathSyntax/simpleGetMessagePathDataItems";
 import { stringifyMessagePath } from "@lichtblick/suite-base/components/MessagePathSyntax/stringifyRosPath";
 import { fillInGlobalVariablesInPath } from "@lichtblick/suite-base/components/MessagePathSyntax/useCachedGetMessagePathDataItems";
+import { UseSubscribeMessageRange } from "@lichtblick/suite-base/components/PanelExtensionAdapter/useSubscribeMessageRange";
 import { InteractionEvent, Scale } from "@lichtblick/suite-base/panels/Plot/types";
 import { PlotXAxisVal } from "@lichtblick/suite-base/panels/Plot/utils/config";
-import { MessageBlock } from "@lichtblick/suite-base/players/types";
 import PlayerBuilder from "@lichtblick/suite-base/testing/builders/PlayerBuilder";
 import PlotBuilder from "@lichtblick/suite-base/testing/builders/PlotBuilder";
 import RosTimeBuilder from "@lichtblick/suite-base/testing/builders/RosTimeBuilder";
@@ -19,7 +19,8 @@ import { BasicBuilder } from "@lichtblick/test-builders";
 
 import { OffscreenCanvasRenderer } from "./OffscreenCanvasRenderer";
 import { PlotCoordinator } from "./PlotCoordinator";
-import { IDatasetsBuilder, SeriesItem } from "./builders/IDatasetsBuilder";
+import { IDatasetsBuilder, SeriesConfigKey, SeriesItem } from "./builders/IDatasetsBuilder";
+import { pathToSubscribePayload } from "./utils/subscription";
 
 jest.mock("./OffscreenCanvasRenderer");
 jest.mock("./builders/IDatasetsBuilder");
@@ -60,10 +61,20 @@ jest.mock("@lichtblick/suite-base/components/MessagePathSyntax/stringifyRosPath"
   stringifyMessagePath: jest.fn(),
 }));
 
+jest.mock("./utils/subscription", () => ({
+  pathToSubscribePayload: jest.fn().mockReturnValue(undefined),
+}));
+
+const mockSubscribeMessageRange = jest.fn();
+jest.mock("@lichtblick/suite-base/components/PanelExtensionAdapter", () => ({
+  useSubscribeMessageRange: () => mockSubscribeMessageRange,
+}));
+
 describe("PlotCoordinator", () => {
   let renderer: jest.Mocked<OffscreenCanvasRenderer>;
   let datasetsBuilder: jest.Mocked<IDatasetsBuilder>;
   let plotCoordinator: PlotCoordinator;
+  let subscribeMessageRange: UseSubscribeMessageRange;
 
   beforeEach(() => {
     const canvas = new OffscreenCanvas(500, 500);
@@ -77,10 +88,10 @@ describe("PlotCoordinator", () => {
       pathsWithMismatchedDataLengths: [],
     });
     datasetsBuilder.setSeries = jest.fn();
-    datasetsBuilder.handleBlocks = jest.fn().mockResolvedValue(undefined);
     datasetsBuilder.getCsvData = jest.fn().mockResolvedValue([]);
 
-    plotCoordinator = new PlotCoordinator(renderer, datasetsBuilder);
+    subscribeMessageRange = mockSubscribeMessageRange;
+    plotCoordinator = new PlotCoordinator(renderer, datasetsBuilder, subscribeMessageRange);
   });
 
   afterEach(() => {
@@ -118,7 +129,7 @@ describe("PlotCoordinator", () => {
 
     it("should return immediately if plotCoordinator is destroyed", () => {
       const state = PlayerBuilder.playerState();
-      jest.spyOn(plotCoordinator as any, "isDestroyed").mockReturnValue(true);
+      plotCoordinator.destroy();
       const handlePlayerStateSpy = jest.spyOn(datasetsBuilder, "handlePlayerState");
       const updateSpy = jest.spyOn(renderer, "update");
 
@@ -196,6 +207,96 @@ describe("PlotCoordinator", () => {
     });
   });
 
+  describe("topic range subscriptions", () => {
+    beforeEach(() => {
+      (pathToSubscribePayload as jest.Mock).mockReturnValue({ topic: "/foo", preloadType: "full" });
+      datasetsBuilder.handleMessageRange = jest.fn();
+      mockSubscribeMessageRange.mockReturnValue(jest.fn());
+    });
+
+    it("groups multiple series with the same topic into a single subscription", () => {
+      // Given
+      const topic = "/foo";
+      plotCoordinator["series"] = [
+        {
+          parsed: { topicName: topic, messagePath: [] },
+          key: "0:receiveTime:/foo.x" as SeriesConfigKey,
+          configIndex: 0,
+          timestampMethod: "receiveTime",
+        },
+        {
+          parsed: { topicName: topic, messagePath: [] },
+          key: "1:receiveTime:/foo.y" as SeriesConfigKey,
+          configIndex: 1,
+          timestampMethod: "receiveTime",
+        },
+      ] as unknown as SeriesItem[];
+      const state = PlayerBuilder.playerState({ activeData: PlayerBuilder.activeData() });
+
+      // When
+      plotCoordinator.handlePlayerState(state);
+
+      // Then
+      expect(mockSubscribeMessageRange).toHaveBeenCalledTimes(1);
+      expect(mockSubscribeMessageRange).toHaveBeenCalledWith(expect.objectContaining({ topic }));
+    });
+
+    it("cancels subscription for a topic removed from series", () => {
+      // Given — first call subscribes /foo
+      const cancelFoo = jest.fn();
+      mockSubscribeMessageRange.mockReturnValue(cancelFoo);
+      plotCoordinator["series"] = [
+        {
+          parsed: { topicName: "/foo", messagePath: [] },
+          key: "0:receiveTime:/foo.val" as SeriesConfigKey,
+          configIndex: 0,
+          timestampMethod: "receiveTime",
+        },
+      ] as unknown as SeriesItem[];
+      const state = PlayerBuilder.playerState({ activeData: PlayerBuilder.activeData() });
+      plotCoordinator.handlePlayerState(state);
+
+      // When — second call with /bar only
+      mockSubscribeMessageRange.mockClear();
+      plotCoordinator["series"] = [
+        {
+          parsed: { topicName: "/bar", messagePath: [] },
+          key: "0:receiveTime:/bar.val" as SeriesConfigKey,
+          configIndex: 0,
+          timestampMethod: "receiveTime",
+        },
+      ] as unknown as SeriesItem[];
+      plotCoordinator.handlePlayerState(state);
+
+      // Then
+      expect(cancelFoo).toHaveBeenCalled();
+      expect(mockSubscribeMessageRange).toHaveBeenCalledWith(
+        expect.objectContaining({ topic: "/bar" }),
+      );
+    });
+
+    it("does not re-subscribe when the same topic and keys are unchanged", () => {
+      // Given
+      plotCoordinator["series"] = [
+        {
+          parsed: { topicName: "/foo", messagePath: [] },
+          key: "0:receiveTime:/foo.val" as SeriesConfigKey,
+          configIndex: 0,
+          timestampMethod: "receiveTime",
+        },
+      ] as unknown as SeriesItem[];
+      const state = PlayerBuilder.playerState({ activeData: PlayerBuilder.activeData() });
+      plotCoordinator.handlePlayerState(state);
+      mockSubscribeMessageRange.mockClear();
+
+      // When — same series, same state
+      plotCoordinator.handlePlayerState(state);
+
+      // Then — no new subscription opened
+      expect(mockSubscribeMessageRange).not.toHaveBeenCalled();
+    });
+  });
+
   describe("destroy", () => {
     it("should set 'destroyed' to true when calling destroy", () => {
       plotCoordinator.destroy();
@@ -205,6 +306,15 @@ describe("PlotCoordinator", () => {
   });
 
   describe("dispatchRender", () => {
+    it("should return immediately if plotCoordinator is destroyed", async () => {
+      plotCoordinator.destroy();
+
+      await plotCoordinator["dispatchRender"]();
+
+      const updateSpy = jest.spyOn(renderer, "update");
+      expect(updateSpy).not.toHaveBeenCalled();
+    });
+
     it("should call 'update' on the renderer when dispatching render", async () => {
       renderer.update.mockResolvedValue({ x: { min: 0, max: 10 }, y: { min: 0, max: 10 } });
 
@@ -280,20 +390,19 @@ describe("PlotCoordinator", () => {
     });
   });
 
-  describe("dispatchBlocks", () => {
-    it("should process and store message blocks correctly", async () => {
-      datasetsBuilder.handleBlocks = jest.fn().mockResolvedValue(undefined);
-      const blocks = [{}] as MessageBlock[];
-      const startTime = RosTimeBuilder.time();
+  describe("handleConfig", () => {
+    it("should return immediately if plotCoordinator is destroyed", () => {
+      const config = PlotBuilder.config({
+        xAxisVal: "timestamp",
+        followingViewWidth: 10,
+        paths: [],
+      });
 
-      await plotCoordinator["dispatchBlocks"](startTime, blocks);
+      plotCoordinator.destroy();
 
-      const handleBlocksSpyOn = jest.spyOn(datasetsBuilder, "handleBlocks");
-      expect(handleBlocksSpyOn).toHaveBeenCalled();
+      plotCoordinator.handleConfig(config, "light", {});
+      expect(plotCoordinator["isTimeseriesPlot"]).toBe(false);
     });
-  });
-
-  describe("handleConfig", () => {
     it("should set isTimeseriesPlot to true when xAxisVal is 'timestamp'", () => {
       const config = PlotBuilder.config({
         xAxisVal: "timestamp",
diff --git a/packages/suite-base/src/panels/Plot/PlotCoordinator.ts b/packages/suite-base/src/panels/Plot/PlotCoordinator.ts
index 36806ced17..0572933110 100644
--- a/packages/suite-base/src/panels/Plot/PlotCoordinator.ts
+++ b/packages/suite-base/src/panels/Plot/PlotCoordinator.ts
@@ -16,11 +16,11 @@ import { Immutable, Time } from "@lichtblick/suite";
 import { simpleGetMessagePathDataItems } from "@lichtblick/suite-base/components/MessagePathSyntax/simpleGetMessagePathDataItems";
 import { stringifyMessagePath } from "@lichtblick/suite-base/components/MessagePathSyntax/stringifyRosPath";
 import { fillInGlobalVariablesInPath } from "@lichtblick/suite-base/components/MessagePathSyntax/useCachedGetMessagePathDataItems";
+import { UseSubscribeMessageRange } from "@lichtblick/suite-base/components/PanelExtensionAdapter/useSubscribeMessageRange";
 import { Bounds1D } from "@lichtblick/suite-base/components/TimeBasedChart/types";
 import { GlobalVariables } from "@lichtblick/suite-base/hooks/useGlobalVariables";
-import { MessageBlock, PlayerState } from "@lichtblick/suite-base/players/types";
+import { PlayerState } from "@lichtblick/suite-base/players/types";
 import { Bounds } from "@lichtblick/suite-base/types/Bounds";
-import delay from "@lichtblick/suite-base/util/delay";
 import { getContrastColor, getLineColor } from "@lichtblick/suite-base/util/plotColors";
 
 import { OffscreenCanvasRenderer } from "./OffscreenCanvasRenderer";
@@ -40,12 +40,13 @@ import {
   UpdateAction,
 } from "./types";
 import { isReferenceLinePlotPathType, PlotConfig } from "./utils/config";
+import { pathToSubscribePayload } from "./utils/subscription";
 
 const replaceUndefinedWithEmptyDataset = (dataset: Dataset | undefined) => dataset ?? { data: [] };
 
 /**
  * PlotCoordinator interfaces commands and updates between the dataset builder and the chart
- * renderer.
+ * renderer. Uses subscribeMessageRange for full dataset history.
  */
 export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
   private renderer: OffscreenCanvasRenderer;
@@ -74,22 +75,39 @@ export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
   private queueDispatchRender = debouncePromise(this.dispatchRender.bind(this));
   private queueDispatchDownsample = debouncePromise(this.dispatchDownsample.bind(this));
   private queueDatasetsRender = debouncePromise(this.dispatchDatasetsRender.bind(this));
-  private queueBlocks = debouncePromise(this.dispatchBlocks.bind(this));
   private destroyed = false;
-  private latestBlocks?: Immutable<(MessageBlock | undefined)[]>;
 
-  public constructor(renderer: OffscreenCanvasRenderer, builder: IDatasetsBuilder) {
+  private readonly subscribeMessageRange: UseSubscribeMessageRange;
+  private readonly rangeSubscriptionCancels = new Map<
+    string,
+    { cancel: () => void; seriesKeys: ReadonlySet<SeriesConfigKey> }
+  >();
+  private startTime: Immutable<Time> | undefined;
+
+  public constructor(
+    renderer: OffscreenCanvasRenderer,
+    builder: IDatasetsBuilder,
+    subscribeMessageRange: UseSubscribeMessageRange,
+  ) {
     super();
-
     this.renderer = renderer;
     this.datasetsBuilder = builder;
+    this.subscribeMessageRange = subscribeMessageRange;
   }
 
   /** Stop the coordinator from sending any future updates to the renderer. */
   public destroy(): void {
+    for (const { cancel } of this.rangeSubscriptionCancels.values()) {
+      cancel();
+    }
+    this.rangeSubscriptionCancels.clear();
     this.destroyed = true;
   }
 
+  private isDestroyed(): boolean {
+    return this.destroyed;
+  }
+
   public setShouldSync({ shouldSync }: { shouldSync: boolean }): void {
     this.shouldSync = shouldSync;
   }
@@ -105,6 +123,25 @@ export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
     }
 
     const { messages, lastSeekTime, currentTime, startTime } = activeData;
+    this.startTime = startTime;
+
+    const seriesKeysByTopic = new Map<string, Set<SeriesConfigKey>>();
+    for (const s of this.series) {
+      if (pathToSubscribePayload(s.parsed, "full") == undefined) {
+        continue;
+      }
+      const keys = seriesKeysByTopic.get(s.parsed.topicName) ?? new Set<SeriesConfigKey>();
+      keys.add(s.key);
+      seriesKeysByTopic.set(s.parsed.topicName, keys);
+    }
+
+    // If the builder uses a separate x-axis topic (e.g. custom x-axis), subscribe to it too.
+    const xTopic = this.datasetsBuilder.getXTopic?.();
+    if (xTopic && !seriesKeysByTopic.has(xTopic)) {
+      seriesKeysByTopic.set(xTopic, new Set<SeriesConfigKey>());
+    }
+
+    this.subscribeTopicRanges(seriesKeysByTopic);
 
     if (this.isTimeseriesPlot) {
       const secondsSinceStart = toSec(subtractTime(currentTime, startTime));
@@ -140,14 +177,6 @@ export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
 
     const handlePlayerStateResult = this.datasetsBuilder.handlePlayerState(state);
 
-    const blocks = state.progress.messageCache?.blocks;
-    if (blocks && this.datasetsBuilder.handleBlocks) {
-      this.latestBlocks = blocks;
-      this.queueBlocks(activeData.startTime, blocks);
-    }
-
-    // There's no result from the builder so we clear dataset range and trigger a render so
-    // we can fall back to other ranges
     if (!handlePlayerStateResult) {
       this.datasetRange = undefined;
       this.queueDispatchRender();
@@ -344,9 +373,6 @@ export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
     return await this.datasetsBuilder.getCsvData();
   }
 
-  /**
-   * Return true if the plot viewport has deviated from the config or dataset bounds and can reset
-   */
   private canReset(): boolean {
     if (this.interactionBounds) {
       return true;
@@ -360,10 +386,6 @@ export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
     return false;
   }
 
-  /**
-   * Get the xBounds if we cleared the interaction and global bounds (i.e) reset
-   * back to the config or dataset bounds
-   */
   private getXResetBounds(): Partial<Bounds1D> {
     const currentSecondsIfFollowMode =
       this.isTimeseriesPlot && this.followRange != undefined && this.currentSeconds != undefined
@@ -381,7 +403,6 @@ export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
   }
 
   private getXBounds(): Partial<Bounds1D> {
-    // Interaction, synced global bounds override the config and data source bounds in precedence
     const resetBounds = this.getXResetBounds();
     return {
       min: this.interactionBounds?.x.min ?? this.globalBounds?.min ?? resetBounds.min,
@@ -389,10 +410,6 @@ export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
     };
   }
 
-  private isDestroyed(): boolean {
-    return this.destroyed;
-  }
-
   private async dispatchRender(): Promise<void> {
     if (this.isDestroyed()) {
       return;
@@ -409,9 +426,7 @@ export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
     const haveInteractionEvents = (this.updateAction.interactionEvents?.length ?? 0) > 0;
 
     const action = this.updateAction;
-    this.updateAction = {
-      type: "update",
-    };
+    this.updateAction = { type: "update" };
 
     const bounds = await this.renderer.update(action);
     if (this.isDestroyed()) {
@@ -469,33 +484,56 @@ export class PlotCoordinator extends EventEmitter<PlotCoordinatorEventTypes> {
     this.emit("xScaleChanged", this.latestXScale);
   }
 
-  private async dispatchBlocks(
-    startTime: Immutable<Time>,
-    blocks: Immutable<(MessageBlock | undefined)[]>,
-  ): Promise<void> {
-    if (!this.datasetsBuilder.handleBlocks) {
+  private cancelTopicSubscription(topic: string): void {
+    this.rangeSubscriptionCancels.get(topic)?.cancel();
+    this.rangeSubscriptionCancels.delete(topic);
+  }
+
+  private subscribeTopicRanges(seriesKeysByTopic: Map<string, Set<SeriesConfigKey>>): void {
+    if (!this.datasetsBuilder.handleMessageRange) {
       return;
     }
+    const builder = this.datasetsBuilder;
 
-    await this.datasetsBuilder.handleBlocks(startTime, blocks, async () => {
-      this.queueDispatchDownsample();
-      // When blocks are fully loaded and a user splits the panel, we are able to process all of the
-      // blocks synchronously. However this creates a poor UX experience for large datasets by
-      // showing nothing on the plot for many seconds while the postMessage prepares a massive send.
-      // This send also hangs the main thread.
-      //
-      // Instead of doing this all synchronously and stalling main thread, we artificially break up
-      // block loading to periodically dispatch the loaded data and render it. This avoids stalling
-      // the main thread and provides visual feedabck to the user that data is loading on the plot.
-      //
-      // await Promise.resolve() does not work as it does not yield enough to the main thread to
-      // dispatch and render.
-      await delay(0);
-
-      // Bail processing if the coordinator has been destroyed or if our input blocks have changed
-      // This lets us start processing new input blocks instead of continuing to work on stale
-      // blocks.
-      return this.isDestroyed() || this.latestBlocks !== blocks;
-    });
+    // Cancel subscriptions for topics that are no longer needed
+    for (const topic of this.rangeSubscriptionCancels.keys()) {
+      if (!seriesKeysByTopic.has(topic)) {
+        this.cancelTopicSubscription(topic);
+      }
+    }
+
+    // Start subscriptions only for new or changed topics
+    for (const [topic, currentKeys] of seriesKeysByTopic) {
+      const existing = this.rangeSubscriptionCancels.get(topic);
+      if (existing) {
+        const keysChanged =
+          existing.seriesKeys.size !== currentKeys.size ||
+          [...currentKeys].some((k) => !existing.seriesKeys.has(k));
+        if (!keysChanged) {
+          continue;
+        }
+        this.cancelTopicSubscription(topic);
+      }
+
+      const cancel = this.subscribeMessageRange({
+        topic,
+        onNewRangeIterator: async (batchIterator) => {
+          let isReset = true;
+          for await (const batch of batchIterator) {
+            if (this.isDestroyed()) {
+              return;
+            }
+            const startTime = this.startTime;
+            if (!startTime) {
+              continue;
+            }
+            builder.handleMessageRange!(batch, { isReset }, startTime);
+            isReset = false;
+            this.queueDispatchDownsample();
+          }
+        },
+      });
+      this.rangeSubscriptionCancels.set(topic, { cancel, seriesKeys: new Set(currentKeys) });
+    }
   }
 }
diff --git a/packages/suite-base/src/panels/Plot/builders/BlockTopicCursor.test.ts b/packages/suite-base/src/panels/Plot/builders/BlockTopicCursor.test.ts
deleted file mode 100644
index 66c151286a..0000000000
--- a/packages/suite-base/src/panels/Plot/builders/BlockTopicCursor.test.ts
+++ /dev/null
@@ -1,202 +0,0 @@
-// SPDX-FileCopyrightText: Copyright (C) 2023-2026 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)<lichtblick@bmwgroup.com>
-// SPDX-License-Identifier: MPL-2.0
-
-// This Source Code Form is subject to the terms of the Mozilla Public
-// License, v2.0. If a copy of the MPL was not distributed with this
-// file, You can obtain one at http://mozilla.org/MPL/2.0/
-
-import * as _ from "lodash-es";
-
-import { MessageEvent } from "@lichtblick/suite";
-
-import { BlockTopicCursor } from "./BlockTopicCursor";
-
-function groupByTopic(events: MessageEvent[]): Record<string, MessageEvent[]> {
-  return _.groupBy(events, (item) => item.topic);
-}
-
-describe("BlockTopicCursor", () => {
-  it("should produce next block through repeated calls to next", () => {
-    const cursor = new BlockTopicCursor("/foo");
-
-    const block0 = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
-        {
-          topic: "/foo",
-          schemaName: "foo",
-          receiveTime: { sec: 0, nsec: 0 },
-          sizeInBytes: 0,
-          message: {
-            val: 1,
-          },
-        },
-      ]),
-    };
-
-    const block1 = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
-        {
-          topic: "/foo",
-          schemaName: "foo",
-          receiveTime: { sec: 0, nsec: 0 },
-          sizeInBytes: 0,
-          message: {
-            val: 2,
-          },
-        },
-      ]),
-    };
-
-    const blocks = [block0, block1];
-
-    expect(cursor.nextWillReset(blocks)).toEqual(true);
-    expect(cursor.next(blocks)).toEqual(block0["messagesByTopic"]["/foo"]);
-    expect(cursor.next(blocks)).toEqual(block1["messagesByTopic"]["/foo"]);
-    expect(cursor.next(blocks)).toEqual(undefined);
-  });
-
-  it("should continue from last location when new blocks arrive", () => {
-    const cursor = new BlockTopicCursor("/foo");
-
-    const block0 = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
-        {
-          topic: "/foo",
-          schemaName: "foo",
-          receiveTime: { sec: 0, nsec: 0 },
-          sizeInBytes: 0,
-          message: {
-            val: 1,
-          },
-        },
-      ]),
-    };
-
-    const block1 = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
-        {
-          topic: "/foo",
-          schemaName: "foo",
-          receiveTime: { sec: 0, nsec: 0 },
-          sizeInBytes: 0,
-          message: {
-            val: 2,
-          },
-        },
-      ]),
-    };
-
-    expect(cursor.nextWillReset([block0])).toEqual(true);
-    expect(cursor.next([block0])).toEqual(block0["messagesByTopic"]["/foo"]);
-    expect(cursor.next([block0])).toEqual(undefined);
-    expect(cursor.next([block0, block1])).toEqual(block1["messagesByTopic"]["/foo"]);
-    expect(cursor.next([block0, block1])).toEqual(undefined);
-  });
-
-  it("should restart if either the first or last iterated block of the topic change", () => {
-    // When using per-field subscriptions, a particular topic might be populated across all blocks
-    // because a field was needed. A cursor will think it has processed all the blocks but a new
-    // subscription update will start to update blocks with the new field. We need to properly
-    // detect this happening - that the block is changed.
-
-    const cursor = new BlockTopicCursor("/foo");
-
-    const block0 = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
-        {
-          topic: "/foo",
-          schemaName: "foo",
-          receiveTime: { sec: 0, nsec: 0 },
-          sizeInBytes: 0,
-          message: {
-            val: 1,
-          },
-        },
-      ]),
-    };
-
-    const block1 = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
-        {
-          topic: "/foo",
-          schemaName: "foo",
-          receiveTime: { sec: 0, nsec: 0 },
-          sizeInBytes: 0,
-          message: {
-            val: 2,
-          },
-        },
-      ]),
-    };
-
-    const block2 = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
-        {
-          topic: "/foo",
-          schemaName: "foo",
-          receiveTime: { sec: 0, nsec: 0 },
-          sizeInBytes: 0,
-          message: {
-            val: 3,
-          },
-        },
-      ]),
-    };
-
-    const blocks = [block0, block1, block2];
-
-    // consume all the blocks because the topic is present in all blocks
-    expect(cursor.nextWillReset(blocks)).toEqual(true);
-    expect(cursor.next(blocks)).toEqual(block0["messagesByTopic"]["/foo"]);
-    expect(cursor.next(blocks)).toEqual(block1["messagesByTopic"]["/foo"]);
-    expect(cursor.next(blocks)).toEqual(block2["messagesByTopic"]["/foo"]);
-    expect(cursor.next(blocks)).toEqual(undefined);
-    expect(cursor.nextWillReset(blocks)).toEqual(false);
-
-    // The first block is changed, we should reset
-    block0.messagesByTopic["/foo"] = [
-      {
-        topic: "/foo",
-        schemaName: "foo",
-        receiveTime: { sec: 0, nsec: 0 },
-        sizeInBytes: 0,
-        message: {
-          val: 1,
-          another: 2,
-        },
-      },
-    ];
-
-    expect(cursor.nextWillReset(blocks)).toEqual(true);
-    expect(cursor.next(blocks)).toEqual(block0["messagesByTopic"]["/foo"]);
-    expect(cursor.next(blocks)).toEqual(block1["messagesByTopic"]["/foo"]);
-
-    // The last block we produced changed, we also need to reset
-    block1.messagesByTopic["/foo"] = [
-      {
-        topic: "/foo",
-        schemaName: "foo",
-        receiveTime: { sec: 0, nsec: 0 },
-        sizeInBytes: 0,
-        message: {
-          val: 2,
-          another: 3,
-        },
-      },
-    ];
-
-    expect(cursor.nextWillReset(blocks)).toEqual(true);
-    expect(cursor.next(blocks)).toEqual(block0["messagesByTopic"]["/foo"]);
-    expect(cursor.next(blocks)).toEqual(block1["messagesByTopic"]["/foo"]);
-    expect(cursor.next(blocks)).toEqual(block2["messagesByTopic"]["/foo"]);
-    expect(cursor.next(blocks)).toEqual(undefined);
-    expect(cursor.nextWillReset(blocks)).toEqual(false);
-  });
-});
diff --git a/packages/suite-base/src/panels/Plot/builders/BlockTopicCursor.ts b/packages/suite-base/src/panels/Plot/builders/BlockTopicCursor.ts
deleted file mode 100644
index 753953fc3d..0000000000
--- a/packages/suite-base/src/panels/Plot/builders/BlockTopicCursor.ts
+++ /dev/null
@@ -1,69 +0,0 @@
-// SPDX-FileCopyrightText: Copyright (C) 2023-2026 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)<lichtblick@bmwgroup.com>
-// SPDX-License-Identifier: MPL-2.0
-
-// This Source Code Form is subject to the terms of the Mozilla Public
-// License, v2.0. If a copy of the MPL was not distributed with this
-// file, You can obtain one at http://mozilla.org/MPL/2.0/
-
-import { Immutable, MessageEvent } from "@lichtblick/suite";
-import { MessageBlock } from "@lichtblick/suite-base/players/types";
-
-/**
- * BlockTopicCursor tracks the last seen block messages for a given topic and can produce the next
- * block that has not yet been processed.
- *
- * When block topic data changes, it re-starts _next_.
- */
-export class BlockTopicCursor {
-  #firstBlockRef: Immutable<MessageEvent[]> | undefined;
-  #lastBlockRef: Immutable<MessageEvent[]> | undefined;
-
-  #nextBlockIdx = 0;
-  #topic: string;
-
-  public constructor(topic: string) {
-    this.#topic = topic;
-  }
-
-  /**
-   * Return true if the cursor is invalidated and reading next will start from the beginning
-   */
-  public nextWillReset(blocks: Immutable<(MessageBlock | undefined)[]>): boolean {
-    const firstBlockRef = blocks[0]?.messagesByTopic[this.#topic];
-
-    const lastIdx = Math.max(0, this.#nextBlockIdx - 1);
-    const lastBlockRef = blocks[lastIdx]?.messagesByTopic[this.#topic];
-
-    return firstBlockRef !== this.#firstBlockRef || lastBlockRef !== this.#lastBlockRef;
-  }
-
-  /**
-   * Given an array of blocks, return the next set of messages.
-   *
-   * When the underlying topic data changes, the cursor is reset.
-   */
-  public next(
-    blocks: Immutable<(MessageBlock | undefined)[]>,
-  ): Immutable<MessageEvent[]> | undefined {
-    if (this.nextWillReset(blocks)) {
-      const firstBlockRef = blocks[0]?.messagesByTopic[this.#topic];
-      this.#nextBlockIdx = 0;
-      this.#firstBlockRef = firstBlockRef;
-    }
-
-    const idx = this.#nextBlockIdx;
-    if (idx >= blocks.length) {
-      return undefined;
-    }
-
-    // if the block is not yet loaded we do not increment next
-    const blockTopic = blocks[idx]?.messagesByTopic[this.#topic];
-    if (blockTopic == undefined) {
-      return undefined;
-    }
-
-    this.#nextBlockIdx += 1;
-    this.#lastBlockRef = blockTopic;
-    return blockTopic;
-  }
-}
diff --git a/packages/suite-base/src/panels/Plot/builders/CurrentCustomDatasetsBuilder.test.ts b/packages/suite-base/src/panels/Plot/builders/CurrentCustomDatasetsBuilder.test.ts
index f9c9edea4b..4d091a5a12 100644
--- a/packages/suite-base/src/panels/Plot/builders/CurrentCustomDatasetsBuilder.test.ts
+++ b/packages/suite-base/src/panels/Plot/builders/CurrentCustomDatasetsBuilder.test.ts
@@ -8,7 +8,6 @@
 import { unwrap } from "@lichtblick/den/monads";
 import { parseMessagePath } from "@lichtblick/message-path";
 import {
-  MessageBlock,
   PlayerPresence,
   PlayerState,
   PlayerStateActiveData,
@@ -40,10 +39,7 @@ function buildSeriesItems(
   });
 }
 
-function buildPlayerState(
-  activeDataOverride?: Partial<PlayerStateActiveData>,
-  blocks?: readonly (MessageBlock | undefined)[],
-): PlayerState {
+function buildPlayerState(activeDataOverride?: Partial<PlayerStateActiveData>): PlayerState {
   return {
     activeData: {
       messages: [],
@@ -65,10 +61,6 @@ function buildPlayerState(
     playerId: "1",
     progress: {
       fullyLoadedFractionRanges: [],
-      messageCache: {
-        blocks: blocks ?? [],
-        startTime: { sec: 0, nsec: 0 },
-      },
     },
   };
 }
diff --git a/packages/suite-base/src/panels/Plot/builders/CurrentCustomDatasetsBuilder.ts b/packages/suite-base/src/panels/Plot/builders/CurrentCustomDatasetsBuilder.ts
index 7cc0f4ab45..bcdf057477 100644
--- a/packages/suite-base/src/panels/Plot/builders/CurrentCustomDatasetsBuilder.ts
+++ b/packages/suite-base/src/panels/Plot/builders/CurrentCustomDatasetsBuilder.ts
@@ -5,12 +5,11 @@
 // License, v2.0. If a copy of the MPL was not distributed with this
 // file, You can obtain one at http://mozilla.org/MPL/2.0/
 
-import { ChartDataset } from "chart.js";
 import * as _ from "lodash-es";
 
 import { filterMap } from "@lichtblick/den/collection";
 import { MessagePath } from "@lichtblick/message-path";
-import { Immutable, Time, MessageEvent } from "@lichtblick/suite";
+import { Immutable } from "@lichtblick/suite";
 import { simpleGetMessagePathDataItems } from "@lichtblick/suite-base/components/MessagePathSyntax/simpleGetMessagePathDataItems";
 import { mathFunctions } from "@lichtblick/suite-base/panels/Plot/utils/mathFunctions";
 import { PlayerState } from "@lichtblick/suite-base/players/types";
@@ -23,20 +22,9 @@ import {
   SeriesConfigKey,
   SeriesItem,
 } from "./IDatasetsBuilder";
-import { Dataset } from "../types";
-import { getChartValue, isChartValue, Datum } from "../utils/datum";
-
-type DatumWithReceiveTime = Datum & {
-  receiveTime: Time;
-};
-
-type CurrentCustomSeriesItem = {
-  configIndex: number;
-  enabled: boolean;
-  messagePath: string;
-  parsed: Immutable<MessagePath>;
-  dataset: ChartDataset<"scatter", DatumWithReceiveTime[]>;
-};
+import { CurrentFrameSeriesItem } from "./types";
+import { buildViewportDatasets, lastMatchingTopic, setSeries } from "./utils";
+import { getChartValue, isChartValue } from "../utils/datum";
 
 /**
  * CurrentCustomDatasetsBuilder builds datasets from a custom x-axis message path and
@@ -47,7 +35,7 @@ export class CurrentCustomDatasetsBuilder implements IDatasetsBuilder {
 
   #xValues: number[] = [];
 
-  #seriesByKey = new Map<SeriesConfigKey, CurrentCustomSeriesItem>();
+  #seriesByKey = new Map<SeriesConfigKey, CurrentFrameSeriesItem>();
   #pathsWithMismatchedDataLengths = new Set<string>();
 
   // Process the latest messages from the player state to extract any updated x or y values
@@ -149,38 +137,7 @@ export class CurrentCustomDatasetsBuilder implements IDatasetsBuilder {
   }
 
   public setSeries(series: Immutable<SeriesItem[]>): void {
-    // Make a new map so we drop series which are no longer present
-    const newSeries = new Map();
-
-    for (const item of series) {
-      let existingSeries = this.#seriesByKey.get(item.key);
-      existingSeries ??= {
-        configIndex: item.configIndex,
-        enabled: item.enabled,
-        messagePath: item.messagePath,
-        parsed: item.parsed,
-        dataset: {
-          data: [],
-        },
-      };
-
-      existingSeries.configIndex = item.configIndex;
-      existingSeries.enabled = item.enabled;
-      existingSeries.dataset = {
-        ...existingSeries.dataset,
-        borderColor: item.color,
-        showLine: item.showLine,
-        fill: false,
-        borderWidth: item.lineSize,
-        pointRadius: item.lineSize * 1.2,
-        pointHoverRadius: 3,
-        pointBackgroundColor: item.showLine ? item.contrastColor : item.color,
-        pointBorderColor: "transparent",
-      };
-
-      newSeries.set(item.key, existingSeries);
-    }
-    this.#seriesByKey = newSeries;
+    this.#seriesByKey = setSeries(this.#seriesByKey, series);
   }
 
   // We don't use the viewport because we do not do any downsampling on the assumption that
@@ -188,17 +145,7 @@ export class CurrentCustomDatasetsBuilder implements IDatasetsBuilder {
   //
   // If that assumption changes then downsampling can be revisited.
   public async getViewportDatasets(): Promise<GetViewportDatasetsResult> {
-    const datasets: Dataset[] = [];
-    for (const series of this.#seriesByKey.values()) {
-      if (series.enabled) {
-        datasets[series.configIndex] = series.dataset;
-      }
-    }
-
-    return {
-      datasetsByConfigIndex: datasets,
-      pathsWithMismatchedDataLengths: this.#pathsWithMismatchedDataLengths,
-    };
+    return buildViewportDatasets(this.#seriesByKey, this.#pathsWithMismatchedDataLengths);
   }
 
   public async getCsvData(): Promise<CsvDataset[]> {
@@ -217,14 +164,3 @@ export class CurrentCustomDatasetsBuilder implements IDatasetsBuilder {
     return datasets;
   }
 }
-
-function lastMatchingTopic(msgEvents: Immutable<MessageEvent[]>, topic: string) {
-  for (let i = msgEvents.length - 1; i >= 0; --i) {
-    const msgEvent = msgEvents[i]!;
-    if (msgEvent.topic === topic) {
-      return msgEvent;
-    }
-  }
-
-  return undefined;
-}
diff --git a/packages/suite-base/src/panels/Plot/builders/CustomDatasetsBuilder.test.ts b/packages/suite-base/src/panels/Plot/builders/CustomDatasetsBuilder.test.ts
index db7d752bca..78c3af9150 100644
--- a/packages/suite-base/src/panels/Plot/builders/CustomDatasetsBuilder.test.ts
+++ b/packages/suite-base/src/panels/Plot/builders/CustomDatasetsBuilder.test.ts
@@ -28,10 +28,6 @@ Object.defineProperty(global, "Worker", {
   value: makeComlinkWorkerMock(() => new CustomDatasetsBuilderImpl()),
 });
 
-function groupByTopic(events: MessageEvent[]): Record<string, MessageEvent[]> {
-  return _.groupBy(events, (item) => item.topic);
-}
-
 function buildSeriesItems(
   paths: (Partial<PlotPath> & { key?: string; value: string })[],
 ): SeriesItem[] {
@@ -209,7 +205,7 @@ describe("CustomDatasetsBuilder", () => {
     });
   });
 
-  it("should build updates from blocks", async () => {
+  it("should build updates from message ranges", async () => {
     const builder = new CustomDatasetsBuilder();
 
     builder.setXPath(parseMessagePath("/foo.val"));
@@ -230,86 +226,83 @@ describe("CustomDatasetsBuilder", () => {
       ]),
     );
 
-    const block0 = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
+    // First range batch: x(0,1), bar(0)
+    builder.handleMessageRange(
+      [
         {
           topic: "/foo",
           schemaName: "foo",
           receiveTime: { sec: 0, nsec: 0 },
           sizeInBytes: 0,
-          message: {
-            val: 0,
-          },
+          message: { val: 0 },
         },
         {
           topic: "/foo",
           schemaName: "foo",
           receiveTime: { sec: 0, nsec: 0 },
           sizeInBytes: 0,
-          message: {
-            val: 1,
-          },
+          message: { val: 1 },
         },
+      ],
+      { isReset: false },
+    );
+    builder.handleMessageRange(
+      [
         {
           topic: "/bar",
           schemaName: "bar",
           receiveTime: { sec: 0, nsec: 0 },
           sizeInBytes: 0,
-          message: {
-            val: 0,
-          },
+          message: { val: 0 },
         },
-      ]),
-    };
-
-    // Baz is empty in the first block
-    block0.messagesByTopic["/baz"] = [];
+      ],
+      { isReset: false },
+    );
 
-    const block1 = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
+    // Second range batch: x(2), bar(1,2), baz(4)
+    builder.handleMessageRange(
+      [
         {
           topic: "/foo",
           schemaName: "foo",
           receiveTime: { sec: 0, nsec: 0 },
           sizeInBytes: 0,
-          message: {
-            val: 2,
-          },
+          message: { val: 2 },
         },
+      ],
+      { isReset: false },
+    );
+    builder.handleMessageRange(
+      [
         {
           topic: "/bar",
           schemaName: "bar",
           receiveTime: { sec: 0, nsec: 0 },
           sizeInBytes: 0,
-          message: {
-            val: 1,
-          },
+          message: { val: 1 },
         },
         {
           topic: "/bar",
           schemaName: "bar",
           receiveTime: { sec: 0, nsec: 0 },
           sizeInBytes: 0,
-          message: {
-            val: 2,
-          },
+          message: { val: 2 },
         },
+      ],
+      { isReset: false },
+    );
+    builder.handleMessageRange(
+      [
         {
           topic: "/baz",
           schemaName: "baz",
           receiveTime: { sec: 0, nsec: 0 },
           sizeInBytes: 0,
-          message: {
-            val: 4,
-          },
+          message: { val: 4 },
         },
-      ]),
-    };
-
-    builder.handlePlayerState(buildPlayerState({}, [block0]));
-    builder.handlePlayerState(buildPlayerState({}, [block0, block1]));
+      ],
+      { isReset: false },
+    );
 
     const result = await builder.getViewportDatasets({
       size: { width: 1_000, height: 1_000 },
@@ -339,103 +332,65 @@ describe("CustomDatasetsBuilder", () => {
     });
   });
 
-  it.each(["current", "blocks"] as const)("combines all values from arrays (%s)", async (type) => {
+  it("should reset full data when isReset is true", async () => {
     const builder = new CustomDatasetsBuilder();
 
-    builder.setXPath(parseMessagePath("/foo.values[:].val"));
+    builder.setXPath(parseMessagePath("/foo.val"));
     builder.setSeries(
-      buildSeriesItems([
+      buildSeriesItems([{ enabled: true, timestampMethod: "receiveTime", value: "/bar.val" }]),
+    );
+
+    // Initial data
+    builder.handleMessageRange(
+      [
         {
-          enabled: true,
-          timestampMethod: "receiveTime",
-          value: "/bar.values[:].val",
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: { val: 0 },
         },
+      ],
+      { isReset: false },
+    );
+    builder.handleMessageRange(
+      [
         {
-          enabled: true,
-          timestampMethod: "receiveTime",
-          value: "/baz.values[:].val",
+          topic: "/bar",
+          schemaName: "bar",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: { val: 10 },
         },
-      ]),
+      ],
+      { isReset: false },
     );
 
-    let latestBlocks: MessageBlock[] = [];
-    const sendMessages = (messages: MessageEvent[]) => {
-      if (type === "current") {
-        builder.handlePlayerState(buildPlayerState({ messages }));
-      } else {
-        latestBlocks = [
-          ...latestBlocks,
-          {
-            sizeInBytes: 0,
-            messagesByTopic: {
-              "/baz": [],
-              ...groupByTopic(messages),
-            },
-          },
-        ];
-        builder.handlePlayerState(buildPlayerState({}, latestBlocks));
-      }
-    };
-
-    sendMessages([
-      {
-        topic: "/foo",
-        schemaName: "foo",
-        receiveTime: { sec: 0, nsec: 0 },
-        sizeInBytes: 0,
-        message: {
-          values: [{ val: 0 }, { val: 1 }, { val: 2 }],
-        },
-      },
-      {
-        topic: "/foo",
-        schemaName: "foo",
-        receiveTime: { sec: 0, nsec: 0 },
-        sizeInBytes: 0,
-        message: {
-          values: [{ val: 3 }],
-        },
-      },
-      {
-        topic: "/bar",
-        schemaName: "bar",
-        receiveTime: { sec: 0, nsec: 0 },
-        sizeInBytes: 0,
-        message: {
-          values: [{ val: 10 }, { val: 11 }],
-        },
-      },
-    ]);
-
-    sendMessages([
-      {
-        topic: "/foo",
-        schemaName: "foo",
-        receiveTime: { sec: 0, nsec: 0 },
-        sizeInBytes: 0,
-        message: {
-          values: [{ val: 4 }],
-        },
-      },
-      {
-        topic: "/bar",
-        schemaName: "bar",
-        receiveTime: { sec: 0, nsec: 0 },
-        sizeInBytes: 0,
-        message: {
-          values: [{ val: 12 }, { val: 13 }, { val: 14 }],
+    // Reset replaces all previous data
+    builder.handleMessageRange(
+      [
+        {
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: { val: 1 },
         },
-      },
-      {
-        topic: "/baz",
-        schemaName: "baz",
-        receiveTime: { sec: 0, nsec: 0 },
-        sizeInBytes: 0,
-        message: {
-          values: [{ val: 20 }, { val: 21 }],
+      ],
+      { isReset: true },
+    );
+    builder.handleMessageRange(
+      [
+        {
+          topic: "/bar",
+          schemaName: "bar",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: { val: 99 },
         },
-      },
-    ]);
+      ],
+      { isReset: true },
+    );
 
     const result = await builder.getViewportDatasets({
       size: { width: 1_000, height: 1_000 },
@@ -443,30 +398,138 @@ describe("CustomDatasetsBuilder", () => {
     });
 
     expect(result).toEqual({
-      pathsWithMismatchedDataLengths: new Set(["/baz.values[:].val"]),
+      pathsWithMismatchedDataLengths: new Set(),
       datasetsByConfigIndex: [
         expect.objectContaining({
-          data: [
-            { x: 0, y: 10, value: 10 },
-            { x: 1, y: 11, value: 11 },
-            { x: 2, y: 12, value: 12 },
-            { x: 3, y: 13, value: 13 },
-            { x: 4, y: 14, value: 14 },
-          ],
-          showLine: true,
-          pointRadius: 1.2,
-          fill: false,
-        }),
-        expect.objectContaining({
-          data: [
-            { x: 0, y: 20, value: 20 },
-            { x: 1, y: 21, value: 21 },
-          ],
+          data: [{ x: 1, y: 99, value: 99 }],
         }),
       ],
     });
   });
 
+  it.each(["current", "message range"] as const)(
+    "combines all values from arrays (%s)",
+    async (type) => {
+      const builder = new CustomDatasetsBuilder();
+
+      builder.setXPath(parseMessagePath("/foo.values[:].val"));
+      builder.setSeries(
+        buildSeriesItems([
+          {
+            enabled: true,
+            timestampMethod: "receiveTime",
+            value: "/bar.values[:].val",
+          },
+          {
+            enabled: true,
+            timestampMethod: "receiveTime",
+            value: "/baz.values[:].val",
+          },
+        ]),
+      );
+
+      const sendMessages = (messages: MessageEvent[]) => {
+        if (type === "current") {
+          builder.handlePlayerState(buildPlayerState({ messages }));
+        } else {
+          const byTopic = _.groupBy(messages, (item) => item.topic);
+          for (const topicMessages of Object.values(byTopic)) {
+            builder.handleMessageRange(topicMessages, { isReset: false });
+          }
+        }
+      };
+
+      sendMessages([
+        {
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: {
+            values: [{ val: 0 }, { val: 1 }, { val: 2 }],
+          },
+        },
+        {
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: {
+            values: [{ val: 3 }],
+          },
+        },
+        {
+          topic: "/bar",
+          schemaName: "bar",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: {
+            values: [{ val: 10 }, { val: 11 }],
+          },
+        },
+      ]);
+
+      sendMessages([
+        {
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: {
+            values: [{ val: 4 }],
+          },
+        },
+        {
+          topic: "/bar",
+          schemaName: "bar",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: {
+            values: [{ val: 12 }, { val: 13 }, { val: 14 }],
+          },
+        },
+        {
+          topic: "/baz",
+          schemaName: "baz",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: {
+            values: [{ val: 20 }, { val: 21 }],
+          },
+        },
+      ]);
+
+      const result = await builder.getViewportDatasets({
+        size: { width: 1_000, height: 1_000 },
+        bounds: {},
+      });
+
+      expect(result).toEqual({
+        pathsWithMismatchedDataLengths: new Set(["/baz.values[:].val"]),
+        datasetsByConfigIndex: [
+          expect.objectContaining({
+            data: [
+              { x: 0, y: 10, value: 10 },
+              { x: 1, y: 11, value: 11 },
+              { x: 2, y: 12, value: 12 },
+              { x: 3, y: 13, value: 13 },
+              { x: 4, y: 14, value: 14 },
+            ],
+            showLine: true,
+            pointRadius: 1.2,
+            fill: false,
+          }),
+          expect.objectContaining({
+            data: [
+              { x: 0, y: 20, value: 20 },
+              { x: 1, y: 21, value: 21 },
+            ],
+          }),
+        ],
+      });
+    },
+  );
+
   it("supports toggling series enabled state", async () => {
     const builder = new CustomDatasetsBuilder();
 
diff --git a/packages/suite-base/src/panels/Plot/builders/CustomDatasetsBuilder.ts b/packages/suite-base/src/panels/Plot/builders/CustomDatasetsBuilder.ts
index 0a5d4198e5..28d32e4798 100644
--- a/packages/suite-base/src/panels/Plot/builders/CustomDatasetsBuilder.ts
+++ b/packages/suite-base/src/panels/Plot/builders/CustomDatasetsBuilder.ts
@@ -14,7 +14,6 @@ import { Bounds1D } from "@lichtblick/suite-base/components/TimeBasedChart/types
 import { PlayerState } from "@lichtblick/suite-base/players/types";
 import { extendBounds1D, unionBounds1D } from "@lichtblick/suite-base/types/Bounds";
 
-import { BlockTopicCursor } from "./BlockTopicCursor";
 import {
   CustomDatasetsBuilderImpl,
   UpdateDataAction,
@@ -28,12 +27,12 @@ import {
   SeriesItem,
   Viewport,
 } from "./IDatasetsBuilder";
+import { buildCurrentSeriesActions, buildFullSeriesActions } from "./utils";
 import { getChartValue, isChartValue } from "../utils/datum";
 import { MathFunction, mathFunctions } from "../utils/mathFunctions";
 
 type CustomDatasetsSeriesItem = {
   config: Immutable<SeriesItem>;
-  blockCursor: BlockTopicCursor;
 };
 
 // If the datasets builder is garbage collected we also need to cleanup the worker
@@ -43,10 +42,8 @@ const registry = new FinalizationRegistry<() => void>((dispose) => {
 });
 
 export class CustomDatasetsBuilder implements IDatasetsBuilder {
+  readonly #datasetsBuilderRemote: Comlink.Remote<Comlink.RemoteObject<CustomDatasetsBuilderImpl>>;
   #xParsedPath?: Immutable<MessagePath>;
-  #xValuesCursor?: BlockTopicCursor;
-
-  #datasetsBuilderRemote: Comlink.Remote<Comlink.RemoteObject<CustomDatasetsBuilderImpl>>;
 
   #pendingDispatch: Immutable<UpdateDataAction>[] = [];
 
@@ -57,6 +54,12 @@ export class CustomDatasetsBuilder implements IDatasetsBuilder {
   #xCurrentBounds?: Bounds1D;
   #xFullBounds?: Bounds1D;
 
+  #hasRangeSource: boolean = false;
+
+  public getXTopic(): string | undefined {
+    return this.#xParsedPath?.topicName;
+  }
+
   public constructor() {
     const worker = new Worker(
       // foxglove-depcheck-used: babel-plugin-transform-import-meta
@@ -81,7 +84,7 @@ export class CustomDatasetsBuilder implements IDatasetsBuilder {
     let datasetsChanged = false;
 
     const msgEvents = activeData.messages;
-    if (msgEvents.length > 0) {
+    if (!this.#hasRangeSource && msgEvents.length > 0) {
       if (didSeek) {
         this.#pendingDispatch.push({
           type: "reset-current-x",
@@ -91,10 +94,7 @@ export class CustomDatasetsBuilder implements IDatasetsBuilder {
 
       // Read the x-axis values
       if (this.#xParsedPath) {
-        const mathFn = this.#xParsedPath.modifier
-          ? mathFunctions[this.#xParsedPath.modifier]
-          : undefined;
-        const pathItems = readMessagePathItems(msgEvents, this.#xParsedPath, mathFn);
+        const pathItems = parseXPathItems(msgEvents, this.#xParsedPath);
 
         this.#pendingDispatch.push({
           type: "append-current-x",
@@ -107,80 +107,16 @@ export class CustomDatasetsBuilder implements IDatasetsBuilder {
         }
       }
 
-      for (const series of this.#series) {
-        const mathFn = series.config.parsed.modifier
-          ? mathFunctions[series.config.parsed.modifier]
-          : undefined;
-        if (didSeek) {
-          this.#pendingDispatch.push({
-            type: "reset-current",
-            series: series.config.key,
-          });
-        }
-
-        const pathItems = readMessagePathItems(msgEvents, series.config.parsed, mathFn);
-        datasetsChanged ||= pathItems.length > 0;
-        this.#pendingDispatch.push({
-          type: "append-current",
-          series: series.config.key,
-          items: pathItems,
-        });
-      }
-    }
-
-    const blocks = state.progress.messageCache?.blocks;
-    if (blocks) {
-      if (this.#xValuesCursor && this.#xParsedPath) {
-        const mathFn = this.#xParsedPath.modifier
-          ? mathFunctions[this.#xParsedPath.modifier]
-          : undefined;
-
-        if (this.#xValuesCursor.nextWillReset(blocks)) {
-          this.#pendingDispatch.push({
-            type: "reset-full-x",
-          });
-        }
-
-        let messageEvents = undefined;
-        while ((messageEvents = this.#xValuesCursor.next(blocks)) != undefined) {
-          const pathItems = readMessagePathItems(messageEvents, this.#xParsedPath, mathFn);
-
-          this.#pendingDispatch.push({
-            type: "append-full-x",
-            items: pathItems,
-          });
-
-          if (pathItems.length > 0) {
-            datasetsChanged = true;
-            this.#xFullBounds = computeBounds(this.#xFullBounds, pathItems);
-          }
-        }
-      }
-
-      for (const series of this.#series) {
-        const mathFn = series.config.parsed.modifier
-          ? mathFunctions[series.config.parsed.modifier]
-          : undefined;
-
-        if (series.blockCursor.nextWillReset(blocks)) {
-          this.#pendingDispatch.push({
-            type: "reset-full",
-            series: series.config.key,
-          });
-        }
-
-        let messageEvents = undefined;
-        while ((messageEvents = series.blockCursor.next(blocks)) != undefined) {
-          const pathItems = readMessagePathItems(messageEvents, series.config.parsed, mathFn);
-
-          datasetsChanged ||= pathItems.length > 0;
-          this.#pendingDispatch.push({
-            type: "append-full",
-            series: series.config.key,
-            items: pathItems,
-          });
-        }
-      }
+      const { actions: seriesActions, datasetsChanged: seriesChanged } = buildCurrentSeriesActions(
+        this.#series,
+        { didSeek },
+        (config) => {
+          const mathFn = config.parsed.modifier ? mathFunctions[config.parsed.modifier] : undefined;
+          return readMessagePathItems(msgEvents, config.parsed, mathFn);
+        },
+      );
+      this.#pendingDispatch.push(...(seriesActions as UpdateDataAction[]));
+      datasetsChanged ||= seriesChanged;
     }
 
     if (!this.#xCurrentBounds) {
@@ -203,17 +139,48 @@ export class CustomDatasetsBuilder implements IDatasetsBuilder {
     };
   }
 
+  public handleMessageRange(
+    messages: Immutable<MessageEvent[]>,
+    options: { isReset: boolean },
+  ): void {
+    this.#hasRangeSource = true;
+    const topic = messages[0]?.topic;
+    if (!topic) {
+      return;
+    }
+
+    const isXBatch = topic === this.#xParsedPath?.topicName;
+
+    if (isXBatch) {
+      if (options.isReset) {
+        this.#pendingDispatch.push({ type: "reset-full-x" });
+        this.#xFullBounds = undefined;
+      }
+      if (this.#xParsedPath) {
+        const pathItems = parseXPathItems(messages, this.#xParsedPath);
+        if (pathItems.length > 0) {
+          this.#xFullBounds = computeBounds(this.#xFullBounds, pathItems);
+          this.#pendingDispatch.push({ type: "append-full-x", items: pathItems });
+        }
+      }
+      return;
+    }
+
+    const actions = buildFullSeriesActions(this.#series, topic, options, (config) => {
+      const mathFn = config.parsed.modifier ? mathFunctions[config.parsed.modifier] : undefined;
+      return readMessagePathItems(messages, config.parsed, mathFn);
+    });
+    this.#pendingDispatch.push(...(actions as UpdateDataAction[]));
+  }
+
   public setXPath(path: Immutable<MessagePath> | undefined): void {
     if (JSON.stringify(path) === JSON.stringify(this.#xParsedPath)) {
       return;
     }
 
     this.#xParsedPath = path;
-    if (this.#xParsedPath) {
-      this.#xValuesCursor = new BlockTopicCursor(this.#xParsedPath.topicName);
-    } else {
-      this.#xValuesCursor = undefined;
-    }
+    this.#xFullBounds = undefined;
+    this.#xCurrentBounds = undefined;
 
     this.#pendingDispatch.push({
       type: "reset-current-x",
@@ -225,13 +192,7 @@ export class CustomDatasetsBuilder implements IDatasetsBuilder {
   }
 
   public setSeries(series: Immutable<SeriesItem[]>): void {
-    this.#series = series.map((item) => {
-      const existing = this.#series.find((existingItem) => existingItem.config.key === item.key);
-      return {
-        config: item,
-        blockCursor: existing?.blockCursor ?? new BlockTopicCursor(item.parsed.topicName),
-      };
-    });
+    this.#series = series.map((item) => ({ config: item }));
 
     this.#pendingDispatch.push({
       type: "update-series-config",
@@ -256,6 +217,14 @@ export class CustomDatasetsBuilder implements IDatasetsBuilder {
   }
 }
 
+function parseXPathItems(
+  messages: Immutable<MessageEvent[]>,
+  xParsedPath: Immutable<MessagePath>,
+): ValueItem[] {
+  const mathFn = xParsedPath.modifier ? mathFunctions[xParsedPath.modifier] : undefined;
+  return readMessagePathItems(messages, xParsedPath, mathFn);
+}
+
 function readMessagePathItems(
   events: Immutable<MessageEvent[]>,
   path: Immutable<MessagePath>,
diff --git a/packages/suite-base/src/panels/Plot/builders/IDatasetsBuilder.ts b/packages/suite-base/src/panels/Plot/builders/IDatasetsBuilder.ts
index 2f2d816fa9..81b8b87f0d 100644
--- a/packages/suite-base/src/panels/Plot/builders/IDatasetsBuilder.ts
+++ b/packages/suite-base/src/panels/Plot/builders/IDatasetsBuilder.ts
@@ -8,9 +8,9 @@
 import { Opaque } from "ts-essentials";
 
 import { MessagePath } from "@lichtblick/message-path";
-import type { Immutable, Time } from "@lichtblick/suite";
+import type { Immutable, Time, MessageEvent } from "@lichtblick/suite";
 import type { Bounds1D } from "@lichtblick/suite-base/components/TimeBasedChart/types";
-import type { MessageBlock, PlayerState } from "@lichtblick/suite-base/players/types";
+import type { PlayerState } from "@lichtblick/suite-base/players/types";
 import { TimestampMethod } from "@lichtblick/suite-base/util/time";
 
 import type { Dataset } from "../types";
@@ -97,21 +97,29 @@ interface IDatasetsBuilder {
   handlePlayerState(state: Immutable<PlayerState>): HandlePlayerStateResult | undefined;
 
   /**
-   * The builder can provide an implementation of this method to handle block data separately from
-   * current frame player state data.
+   * The builder can provide an implementation of this method to incrementally accumulate full
+   * message history delivered via `subscribeMessageRange`. Each call provides a batch of messages
+   * for a single topic, together with a `startTime` used to compute relative x-axis offsets.
    *
-   * The method is provided a _progress_ callback to call when there is an opportunity to render
-   * some of the processed block data to provide feedback to the caller that work has happened. The
-   * progress callback returns false when further processing should stop.
+   * When `options.isReset` is true the builder should discard any previously accumulated data for
+   * that topic before processing the new batch, signalling the start of a fresh range subscription
+   * (e.g. after a seek). Subsequent calls with `isReset: false` append to the existing data.
    */
-  handleBlocks?(
-    startTime: Immutable<Time>,
-    blocks: Immutable<(MessageBlock | undefined)[]>,
-    progress: () => Promise<boolean>,
-  ): Promise<void>;
+  handleMessageRange?(
+    messages: Immutable<MessageEvent[]>,
+    options: { isReset: boolean },
+    startTime?: Immutable<Time>,
+  ): void;
 
   setSeries(series: Immutable<SeriesItem[]>): void;
 
+  /**
+   * Optional: return the x-axis topic name if this builder uses a separate x-axis topic
+   * (e.g. custom x-axis mode). The coordinator will subscribe to this topic in addition to
+   * the y-series topics so that handleMessageRange receives batches for it.
+   */
+  getXTopic?(): string | undefined;
+
   getViewportDatasets(viewport: Immutable<Viewport>): Promise<GetViewportDatasetsResult>;
 
   getCsvData(): Promise<CsvDataset[]>;
diff --git a/packages/suite-base/src/panels/Plot/builders/IndexDatasetsBuilder.test.ts b/packages/suite-base/src/panels/Plot/builders/IndexDatasetsBuilder.test.ts
index 5cba24f939..38e1c6c1b8 100644
--- a/packages/suite-base/src/panels/Plot/builders/IndexDatasetsBuilder.test.ts
+++ b/packages/suite-base/src/panels/Plot/builders/IndexDatasetsBuilder.test.ts
@@ -8,7 +8,6 @@
 import { unwrap } from "@lichtblick/den/monads";
 import { parseMessagePath } from "@lichtblick/message-path";
 import {
-  MessageBlock,
   PlayerPresence,
   PlayerState,
   PlayerStateActiveData,
@@ -40,10 +39,7 @@ function buildSeriesItems(
   });
 }
 
-function buildPlayerState(
-  activeDataOverride?: Partial<PlayerStateActiveData>,
-  blocks?: readonly (MessageBlock | undefined)[],
-): PlayerState {
+function buildPlayerState(activeDataOverride?: Partial<PlayerStateActiveData>): PlayerState {
   return {
     activeData: {
       messages: [],
@@ -65,10 +61,6 @@ function buildPlayerState(
     playerId: "1",
     progress: {
       fullyLoadedFractionRanges: [],
-      messageCache: {
-        blocks: blocks ?? [],
-        startTime: { sec: 0, nsec: 0 },
-      },
     },
   };
 }
diff --git a/packages/suite-base/src/panels/Plot/builders/IndexDatasetsBuilder.ts b/packages/suite-base/src/panels/Plot/builders/IndexDatasetsBuilder.ts
index 4b99f18ad3..e1970ab9b1 100644
--- a/packages/suite-base/src/panels/Plot/builders/IndexDatasetsBuilder.ts
+++ b/packages/suite-base/src/panels/Plot/builders/IndexDatasetsBuilder.ts
@@ -5,11 +5,8 @@
 // License, v2.0. If a copy of the MPL was not distributed with this
 // file, You can obtain one at http://mozilla.org/MPL/2.0/
 
-import { ChartDataset } from "chart.js";
-
 import { filterMap } from "@lichtblick/den/collection";
-import { MessagePath } from "@lichtblick/message-path";
-import { Immutable, Time, MessageEvent } from "@lichtblick/suite";
+import { Immutable } from "@lichtblick/suite";
 import { simpleGetMessagePathDataItems } from "@lichtblick/suite-base/components/MessagePathSyntax/simpleGetMessagePathDataItems";
 import { Bounds1D } from "@lichtblick/suite-base/components/TimeBasedChart/types";
 import { PlayerState } from "@lichtblick/suite-base/players/types";
@@ -22,26 +19,13 @@ import {
   SeriesConfigKey,
   SeriesItem,
 } from "./IDatasetsBuilder";
+import { CurrentFrameSeriesItem } from "./types";
+import { buildViewportDatasets, lastMatchingTopic, setSeries } from "./utils";
 import { MATH_FUNCTIONS } from "../constants";
-import { Dataset } from "../types";
-import { getChartValue, isChartValue, Datum } from "../utils/datum";
-
-type DatumWithReceiveTime = Datum & {
-  receiveTime: Time;
-};
-
-type IndexDatasetsSeries = {
-  configIndex: number;
-  enabled: boolean;
-  messagePath: string;
-  parsed: Immutable<MessagePath>;
-  dataset: ChartDataset<"scatter", DatumWithReceiveTime[]>;
-};
-
-const emptyPaths = new Set<string>();
+import { getChartValue, isChartValue } from "../utils/datum";
 
 export class IndexDatasetsBuilder implements IDatasetsBuilder {
-  #seriesByKey = new Map<SeriesConfigKey, IndexDatasetsSeries>();
+  #seriesByKey = new Map<SeriesConfigKey, CurrentFrameSeriesItem>();
 
   #range?: Bounds1D;
 
@@ -104,38 +88,7 @@ export class IndexDatasetsBuilder implements IDatasetsBuilder {
   }
 
   public setSeries(series: Immutable<SeriesItem[]>): void {
-    // Make a new map so we drop series which are no longer present
-    const newSeries = new Map();
-
-    for (const item of series) {
-      let existingSeries = this.#seriesByKey.get(item.key);
-      existingSeries ??= {
-        configIndex: item.configIndex,
-        enabled: item.enabled,
-        messagePath: item.messagePath,
-        parsed: item.parsed,
-        dataset: {
-          data: [],
-        },
-      };
-
-      existingSeries.configIndex = item.configIndex;
-      existingSeries.enabled = item.enabled;
-      existingSeries.dataset = {
-        ...existingSeries.dataset,
-        borderColor: item.color,
-        showLine: item.showLine,
-        fill: false,
-        borderWidth: item.lineSize,
-        pointRadius: item.lineSize * 1.2,
-        pointHoverRadius: 3,
-        pointBackgroundColor: item.showLine ? item.contrastColor : item.color,
-        pointBorderColor: "transparent",
-      };
-
-      newSeries.set(item.key, existingSeries);
-    }
-    this.#seriesByKey = newSeries;
+    this.#seriesByKey = setSeries(this.#seriesByKey, series);
   }
 
   // We don't use the viewport because we do not do any downsampling on the assumption that
@@ -143,14 +96,7 @@ export class IndexDatasetsBuilder implements IDatasetsBuilder {
   //
   // If that assumption changes then downsampling can be revisited.
   public async getViewportDatasets(): Promise<GetViewportDatasetsResult> {
-    const datasets: Dataset[] = [];
-    for (const series of this.#seriesByKey.values()) {
-      if (series.enabled) {
-        datasets[series.configIndex] = series.dataset;
-      }
-    }
-
-    return { datasetsByConfigIndex: datasets, pathsWithMismatchedDataLengths: emptyPaths };
+    return buildViewportDatasets(this.#seriesByKey);
   }
 
   public async getCsvData(): Promise<CsvDataset[]> {
@@ -169,14 +115,3 @@ export class IndexDatasetsBuilder implements IDatasetsBuilder {
     return datasets;
   }
 }
-
-function lastMatchingTopic(msgEvents: Immutable<MessageEvent[]>, topic: string) {
-  for (let i = msgEvents.length - 1; i >= 0; --i) {
-    const msgEvent = msgEvents[i]!;
-    if (msgEvent.topic === topic) {
-      return msgEvent;
-    }
-  }
-
-  return undefined;
-}
diff --git a/packages/suite-base/src/panels/Plot/builders/TimestampDatasetsBuilder.test.ts b/packages/suite-base/src/panels/Plot/builders/TimestampDatasetsBuilder.test.ts
index 8e72e6f25b..5d8ff60aab 100644
--- a/packages/suite-base/src/panels/Plot/builders/TimestampDatasetsBuilder.test.ts
+++ b/packages/suite-base/src/panels/Plot/builders/TimestampDatasetsBuilder.test.ts
@@ -5,12 +5,9 @@
 // License, v2.0. If a copy of the MPL was not distributed with this
 // file, You can obtain one at http://mozilla.org/MPL/2.0/
 
-import * as _ from "lodash-es";
-
 import { unwrap } from "@lichtblick/den/monads";
 import { makeComlinkWorkerMock } from "@lichtblick/den/testing";
 import { parseMessagePath } from "@lichtblick/message-path";
-import { MessageEvent } from "@lichtblick/suite";
 import {
   MessageBlock,
   PlayerPresence,
@@ -28,10 +25,6 @@ Object.defineProperty(global, "Worker", {
   value: makeComlinkWorkerMock(() => new TimestampDatasetsBuilderImpl()),
 });
 
-function groupByTopic(events: MessageEvent[]): Record<string, MessageEvent[]> {
-  return _.groupBy(events, (item) => item.topic);
-}
-
 function buildSeriesItems(
   paths: (Partial<PlotPath> & { key?: string; value: string })[],
 ): SeriesItem[] {
@@ -177,61 +170,49 @@ describe("TimestampDatasetsBuilder", () => {
       ]),
     );
 
-    const block = {
-      sizeInBytes: 0,
-      messagesByTopic: groupByTopic([
+    const playerState = buildPlayerState({
+      messages: [
         {
           topic: "/foo",
           schemaName: "foo",
-          receiveTime: { sec: 0, nsec: 0 },
+          receiveTime: { sec: 1, nsec: 0 },
           sizeInBytes: 0,
           message: {
-            val: 0,
+            val: 1.5,
           },
         },
         {
           topic: "/foo",
           schemaName: "foo",
-          receiveTime: { sec: 0.5, nsec: 0 },
+          receiveTime: { sec: 2, nsec: 0 },
           sizeInBytes: 0,
           message: {
-            val: 1,
+            val: 2.5,
           },
         },
-      ]),
-    };
-
-    const playerState = buildPlayerState(
-      {
-        messages: [
-          {
-            topic: "/foo",
-            schemaName: "foo",
-            receiveTime: { sec: 1, nsec: 0 },
-            sizeInBytes: 0,
-            message: {
-              val: 1.5,
-            },
-          },
-          {
-            topic: "/foo",
-            schemaName: "foo",
-            receiveTime: { sec: 2, nsec: 0 },
-            sizeInBytes: 0,
-            message: {
-              val: 2.5,
-            },
-          },
-        ],
-      },
-      [block],
-    );
+      ],
+    });
 
     builder.handlePlayerState(playerState);
-    await builder.handleBlocks(
-      unwrap(playerState.activeData?.startTime),
-      unwrap(playerState.progress.messageCache?.blocks),
-      async () => await Promise.resolve(false),
+    builder.handleMessageRange(
+      [
+        {
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: { val: 0 },
+        },
+        {
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 0.5, nsec: 0 },
+          sizeInBytes: 0,
+          message: { val: 1 },
+        },
+      ],
+      { isReset: false },
+      { sec: 0, nsec: 0 },
     );
 
     await expect(
@@ -255,6 +236,65 @@ describe("TimestampDatasetsBuilder", () => {
     });
   });
 
+  it("should reset full data when isReset is true", async () => {
+    const builder = new TimestampDatasetsBuilder();
+
+    builder.setSeries(
+      buildSeriesItems([{ enabled: true, timestampMethod: "receiveTime", value: "/foo.val" }]),
+    );
+
+    // Initial full data
+    builder.handleMessageRange(
+      [
+        {
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 0, nsec: 0 },
+          sizeInBytes: 0,
+          message: { val: 0 },
+        },
+        {
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 1, nsec: 0 },
+          sizeInBytes: 0,
+          message: { val: 1 },
+        },
+      ],
+      { isReset: false },
+      { sec: 0, nsec: 0 },
+    );
+
+    // Reset replaces all previous full data
+    builder.handleMessageRange(
+      [
+        {
+          topic: "/foo",
+          schemaName: "foo",
+          receiveTime: { sec: 2, nsec: 0 },
+          sizeInBytes: 0,
+          message: { val: 99 },
+        },
+      ],
+      { isReset: true },
+      { sec: 0, nsec: 0 },
+    );
+
+    await expect(
+      builder.getViewportDatasets({
+        size: { width: 1_000, height: 1_000 },
+        bounds: {},
+      }),
+    ).resolves.toEqual({
+      pathsWithMismatchedDataLengths: new Set(),
+      datasetsByConfigIndex: [
+        expect.objectContaining({
+          data: [{ x: 2, y: 99, value: 99 }],
+        }),
+      ],
+    });
+  });
+
   it("computes derivative inside and outside of viewport", async () => {
     const builder = new TimestampDatasetsBuilder();
 
diff --git a/packages/suite-base/src/panels/Plot/builders/TimestampDatasetsBuilder.ts b/packages/suite-base/src/panels/Plot/builders/TimestampDatasetsBuilder.ts
index cbd1b05cf4..39dbee4c92 100644
--- a/packages/suite-base/src/panels/Plot/builders/TimestampDatasetsBuilder.ts
+++ b/packages/suite-base/src/panels/Plot/builders/TimestampDatasetsBuilder.ts
@@ -11,10 +11,9 @@ import { MessagePath } from "@lichtblick/message-path";
 import { toSec, subtract as subtractTime } from "@lichtblick/rostime";
 import { Immutable, MessageEvent, Time } from "@lichtblick/suite";
 import { simpleGetMessagePathDataItems } from "@lichtblick/suite-base/components/MessagePathSyntax/simpleGetMessagePathDataItems";
-import { MessageBlock, PlayerState } from "@lichtblick/suite-base/players/types";
+import { PlayerState } from "@lichtblick/suite-base/players/types";
 import { TimestampMethod, getTimestampForMessage } from "@lichtblick/suite-base/util/time";
 
-import { BlockTopicCursor } from "./BlockTopicCursor";
 import {
   CsvDataset,
   GetViewportDatasetsResult,
@@ -28,6 +27,7 @@ import type {
   TimestampDatasetsBuilderImpl,
   UpdateDataAction,
 } from "./TimestampDatasetsBuilderImpl";
+import { buildCurrentSeriesActions, buildFullSeriesActions } from "./utils";
 import { MATH_FUNCTIONS } from "../constants";
 import { getChartValue, isChartValue } from "../utils/datum";
 import { MathFunction } from "../utils/mathFunctions";
@@ -42,7 +42,6 @@ const emptyPaths = new Set<string>();
 
 type TimestampSeriesItem = {
   config: Immutable<SeriesItem>;
-  blockCursor: BlockTopicCursor;
 };
 
 /**
@@ -60,6 +59,9 @@ export class TimestampDatasetsBuilder implements IDatasetsBuilder {
   #lastSeekTime = 0;
 
   #series: Immutable<TimestampSeriesItem[]> = [];
+  // True once handleMessageRange has been called — indicates a range-capable source (file/bag).
+  // When true, handlePlayerState skips append-current since the full history comes from the range.
+  #hasRangeSource = false;
 
   public constructor() {
     const worker = new Worker(
@@ -84,34 +86,25 @@ export class TimestampDatasetsBuilder implements IDatasetsBuilder {
 
     const msgEvents = activeData.messages;
     let datasetsChanged = false;
-    if (msgEvents.length > 0) {
-      for (const series of this.#series) {
-        const mathFn = series.config.parsed.modifier
-          ? MATH_FUNCTIONS[series.config.parsed.modifier]
-          : undefined;
-
-        if (didSeek) {
-          this.#pendingDispatch.push({
-            type: "reset-current",
-            series: series.config.key,
-          });
-        }
-
-        const pathItems = readMessagePathItems(
-          msgEvents,
-          series.config.parsed,
-          series.config.timestampMethod,
-          activeData.startTime,
-          mathFn,
-        );
-
-        datasetsChanged ||= pathItems.length > 0;
-        this.#pendingDispatch.push({
-          type: "append-current",
-          series: series.config.key,
-          items: pathItems,
-        });
-      }
+    if (!this.#hasRangeSource && msgEvents.length > 0) {
+      const { actions: seriesActions, datasetsChanged: seriesChanged } = buildCurrentSeriesActions(
+        this.#series,
+        { didSeek },
+        (config) => {
+          const mathFn = config.parsed.modifier
+            ? MATH_FUNCTIONS[config.parsed.modifier]
+            : undefined;
+          return readMessagePathItems(
+            msgEvents,
+            config.parsed,
+            config.timestampMethod,
+            activeData.startTime,
+            mathFn,
+          );
+        },
+      );
+      this.#pendingDispatch.push(...(seriesActions as UpdateDataAction[]));
+      datasetsChanged ||= seriesChanged;
     }
 
     return {
@@ -120,76 +113,32 @@ export class TimestampDatasetsBuilder implements IDatasetsBuilder {
     };
   }
 
-  public async handleBlocks(
+  public handleMessageRange(
+    messages: Immutable<MessageEvent[]>,
+    options: { isReset: boolean },
     startTime: Immutable<Time>,
-    blocks: Immutable<(MessageBlock | undefined)[]>,
-    progress: () => Promise<boolean>,
-  ): Promise<void> {
-    // identify if series need resetting because
-    for (const series of this.#series) {
-      if (series.blockCursor.nextWillReset(blocks)) {
-        this.#pendingDispatch.push({
-          type: "reset-full",
-          series: series.config.key,
-        });
-      }
+  ): void {
+    this.#hasRangeSource = true;
+    const topic = messages[0]?.topic;
+    if (!topic) {
+      return;
     }
 
-    const seriesArr = this.#series;
-
-    // We loop through the series and only process one next block and keep doing this until
-    // there are no more updates. This processes the series "in parallel" so that all of them appear
-    // to be loading blocks at the same time.
-    let done = 0;
-    do {
-      done = 0;
-
-      for (const series of seriesArr) {
-        const mathFn = series.config.parsed.modifier
-          ? MATH_FUNCTIONS[series.config.parsed.modifier]
-          : undefined;
-
-        const messageEvents = series.blockCursor.next(blocks);
-        if (!messageEvents) {
-          done += 1;
-          continue;
-        }
-
-        const pathItems = readMessagePathItems(
-          messageEvents,
-          series.config.parsed,
-          series.config.timestampMethod,
-          startTime,
-          mathFn,
-        );
-
-        if (pathItems.length === 0) {
-          continue;
-        }
-
-        this.#pendingDispatch.push({
-          type: "append-full",
-          series: series.config.key,
-          items: pathItems,
-        });
-
-        const abort = await progress();
-        if (abort) {
-          return;
-        }
-      }
-    } while (done < seriesArr.length);
+    const actions = buildFullSeriesActions(this.#series, topic, options, (config) => {
+      const mathFn = config.parsed.modifier ? MATH_FUNCTIONS[config.parsed.modifier] : undefined;
+      return readMessagePathItems(
+        messages,
+        config.parsed,
+        config.timestampMethod,
+        startTime,
+        mathFn,
+      );
+    });
+    this.#pendingDispatch.push(...(actions as UpdateDataAction[]));
   }
 
   public setSeries(series: Immutable<SeriesItem[]>): void {
-    this.#series = series.map((item) => {
-      const existing = this.#series.find((existingItem) => existingItem.config.key === item.key);
-      return {
-        config: item,
-        blockCursor: existing?.blockCursor ?? new BlockTopicCursor(item.parsed.topicName),
-      };
-    });
-
+    this.#series = series.map((item) => ({ config: item }));
     this.#pendingDispatch.push({
       type: "update-series-config",
       seriesItems: series,
diff --git a/packages/suite-base/src/panels/Plot/builders/types.ts b/packages/suite-base/src/panels/Plot/builders/types.ts
new file mode 100644
index 0000000000..796a08dae2
--- /dev/null
+++ b/packages/suite-base/src/panels/Plot/builders/types.ts
@@ -0,0 +1,25 @@
+// SPDX-FileCopyrightText: Copyright (C) 2023-2026 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)<lichtblick@bmwgroup.com>
+// SPDX-License-Identifier: MPL-2.0
+
+import { ChartDataset } from "chart.js";
+
+import { MessagePath } from "@lichtblick/message-path";
+import { Time } from "@lichtblick/rostime";
+import { Immutable } from "@lichtblick/suite";
+import { Datum } from "@lichtblick/suite-base/panels/Plot/utils/datum";
+
+type DatumWithReceiveTime = Datum & {
+  receiveTime: Time;
+};
+
+/**
+ * Series item type shared by builders that work only with the current frame (no accumulation).
+ * Used by IndexDatasetsBuilder and CurrentCustomDatasetsBuilder.
+ */
+export type CurrentFrameSeriesItem = {
+  configIndex: number;
+  enabled: boolean;
+  messagePath: string;
+  parsed: Immutable<MessagePath>;
+  dataset: ChartDataset<"scatter", DatumWithReceiveTime[]>;
+};
diff --git a/packages/suite-base/src/panels/Plot/builders/utils.test.ts b/packages/suite-base/src/panels/Plot/builders/utils.test.ts
new file mode 100644
index 0000000000..31bc1f4263
--- /dev/null
+++ b/packages/suite-base/src/panels/Plot/builders/utils.test.ts
@@ -0,0 +1,198 @@
+// SPDX-FileCopyrightText: Copyright (C) 2023-2026 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)<lichtblick@bmwgroup.com>
+// SPDX-License-Identifier: MPL-2.0
+
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/
+
+import { unwrap } from "@lichtblick/den/monads";
+import { parseMessagePath } from "@lichtblick/message-path";
+import { Immutable, MessageEvent } from "@lichtblick/suite";
+import { BasicBuilder } from "@lichtblick/test-builders";
+
+import { SeriesConfigKey, SeriesItem } from "./IDatasetsBuilder";
+import { buildCurrentSeriesActions, buildFullSeriesActions, lastMatchingTopic } from "./utils";
+
+function makeSeriesItem({
+  key,
+  value,
+  ...rest
+}: { key: string; value: string } & Partial<SeriesItem>): { config: SeriesItem } {
+  const parsed = unwrap(parseMessagePath(value));
+  return {
+    config: {
+      configIndex: 0,
+      parsed,
+      color: "red",
+      contrastColor: "blue",
+      enabled: true,
+      timestampMethod: "receiveTime",
+      key,
+      lineSize: 1,
+      messagePath: value,
+      showLine: true,
+      ...rest,
+    },
+  };
+}
+
+function makeMessageEvent(topic: string, receiveTimeSec = 0): MessageEvent {
+  return {
+    topic,
+    receiveTime: { sec: receiveTimeSec, nsec: 0 },
+    message: {},
+    schemaName: "",
+    sizeInBytes: 0,
+  };
+}
+
+describe("lastMatchingTopic", () => {
+  const matchTopic = BasicBuilder.string();
+  const otherTopic = BasicBuilder.string();
+  it("returns undefined for an empty array", () => {
+    expect(lastMatchingTopic([], matchTopic)).toBeUndefined();
+  });
+
+  it("returns undefined when no event matches the topic", () => {
+    const events = [makeMessageEvent(otherTopic), makeMessageEvent(otherTopic)];
+    expect(lastMatchingTopic(events, matchTopic)).toBeUndefined();
+  });
+
+  it("returns the last matching event", () => {
+    const first = makeMessageEvent(matchTopic, 1);
+    const second = makeMessageEvent(matchTopic, 2);
+    const other = makeMessageEvent(otherTopic, 3);
+    expect(lastMatchingTopic([first, second, other], matchTopic)).toBe(second);
+  });
+
+  it("ignores non-matching events that appear after the match", () => {
+    const match = makeMessageEvent(matchTopic, 1);
+    const after = makeMessageEvent(otherTopic, 2);
+    expect(lastMatchingTopic([match, after], matchTopic)).toBe(match);
+  });
+});
+
+describe("buildCurrentSeriesActions", () => {
+  const seriesA = makeSeriesItem({ key: "a" as SeriesConfigKey, value: "/topic_a.field" });
+  const seriesB = makeSeriesItem({ key: "b" as SeriesConfigKey, value: "/topic_b.field" });
+
+  it("returns empty actions and no change when there are no series", () => {
+    const result = buildCurrentSeriesActions([], { didSeek: false }, () => [1]);
+    expect(result.actions).toEqual([]);
+    expect(result.datasetsChanged).toBe(false);
+  });
+
+  it("emits only append-current actions when didSeek is false", () => {
+    const items = [42];
+    const result = buildCurrentSeriesActions([seriesA], { didSeek: false }, () => items);
+    expect(result.actions).toEqual([{ type: "append-current", series: "a", items }]);
+    expect(result.datasetsChanged).toBe(true);
+  });
+
+  it("emits reset-current before append-current when didSeek is true", () => {
+    const items = [1, 2];
+    const result = buildCurrentSeriesActions([seriesA], { didSeek: true }, () => items);
+    expect(result.actions).toEqual([
+      { type: "reset-current", series: "a" },
+      { type: "append-current", series: "a", items },
+    ]);
+    expect(result.datasetsChanged).toBe(true);
+  });
+
+  it("handles multiple series, emitting actions in order", () => {
+    const result = buildCurrentSeriesActions([seriesA, seriesB], { didSeek: false }, (config) =>
+      config.key === "a" ? [1] : [2, 3],
+    );
+    expect(result.actions).toEqual([
+      { type: "append-current", series: "a", items: [1] },
+      { type: "append-current", series: "b", items: [2, 3] },
+    ]);
+    expect(result.datasetsChanged).toBe(true);
+  });
+
+  it("sets datasetsChanged to false when all series return empty items", () => {
+    const result = buildCurrentSeriesActions([seriesA, seriesB], { didSeek: false }, () => []);
+    expect(result.datasetsChanged).toBe(false);
+    expect(result.actions).toEqual([
+      { type: "append-current", series: "a", items: [] },
+      { type: "append-current", series: "b", items: [] },
+    ]);
+  });
+
+  it("sets datasetsChanged to true when at least one series has items", () => {
+    const result = buildCurrentSeriesActions([seriesA, seriesB], { didSeek: false }, (config) =>
+      config.key === "b" ? [99] : [],
+    );
+    expect(result.datasetsChanged).toBe(true);
+  });
+
+  it("passes the series config to getItems", () => {
+    const captured: Immutable<SeriesItem>[] = [];
+    buildCurrentSeriesActions([seriesA, seriesB], { didSeek: false }, (config) => {
+      captured.push(config);
+      return [];
+    });
+    expect(captured).toEqual([seriesA.config, seriesB.config]);
+  });
+
+  it("emits reset-current for every series when didSeek is true with multiple series", () => {
+    const result = buildCurrentSeriesActions([seriesA, seriesB], { didSeek: true }, () => []);
+    expect(result.actions).toEqual([
+      { type: "reset-current", series: "a" },
+      { type: "append-current", series: "a", items: [] },
+      { type: "reset-current", series: "b" },
+      { type: "append-current", series: "b", items: [] },
+    ]);
+  });
+});
+
+describe("buildFullSeriesActions", () => {
+  const topic = `/${BasicBuilder.string()}`;
+  const otherTopic = `/${BasicBuilder.string()}`;
+  const seriesA = makeSeriesItem({ key: "a" as SeriesConfigKey, value: `${topic}.field` });
+  const seriesB = makeSeriesItem({ key: "b" as SeriesConfigKey, value: `${otherTopic}.field` });
+
+  it("returns empty actions when there are no series", () => {
+    const result = buildFullSeriesActions([], topic, { isReset: false }, () => [1]);
+    expect(result).toEqual([]);
+  });
+
+  it("emits only append-full for a matching topic with items", () => {
+    const items = [42];
+    const result = buildFullSeriesActions([seriesA], topic, { isReset: false }, () => items);
+    expect(result).toEqual([{ type: "append-full", series: "a", items }]);
+  });
+
+  it("emits reset-full before append-full when isReset is true", () => {
+    const items = [1, 2];
+    const result = buildFullSeriesActions([seriesA], topic, { isReset: true }, () => items);
+    expect(result).toEqual([
+      { type: "reset-full", series: "a" },
+      { type: "append-full", series: "a", items },
+    ]);
+  });
+
+  it("emits only reset-full when isReset is true and getItems returns empty", () => {
+    const result = buildFullSeriesActions([seriesA], topic, { isReset: true }, () => []);
+    expect(result).toEqual([{ type: "reset-full", series: "a" }]);
+  });
+
+  it("skips series whose topic does not match", () => {
+    const result = buildFullSeriesActions([seriesB], topic, { isReset: false }, () => [1]);
+    expect(result).toEqual([]);
+  });
+
+  it("passes the series config to getItems", () => {
+    const captured: Immutable<SeriesItem>[] = [];
+    buildFullSeriesActions([seriesA, seriesB], topic, { isReset: false }, (config) => {
+      captured.push(config);
+      return [];
+    });
+    expect(captured).toEqual([seriesA.config]);
+  });
+
+  it("emits no append-full when getItems returns empty and isReset is false", () => {
+    const result = buildFullSeriesActions([seriesA], topic, { isReset: false }, () => []);
+    expect(result).toEqual([]);
+  });
+});
diff --git a/packages/suite-base/src/panels/Plot/builders/utils.ts b/packages/suite-base/src/panels/Plot/builders/utils.ts
new file mode 100644
index 0000000000..6e292f787c
--- /dev/null
+++ b/packages/suite-base/src/panels/Plot/builders/utils.ts
@@ -0,0 +1,146 @@
+// SPDX-FileCopyrightText: Copyright (C) 2023-2026 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)<lichtblick@bmwgroup.com>
+// SPDX-License-Identifier: MPL-2.0
+
+import { Immutable, MessageEvent } from "@lichtblick/suite";
+
+import { GetViewportDatasetsResult, SeriesConfigKey, SeriesItem } from "./IDatasetsBuilder";
+import { CurrentFrameSeriesItem } from "./types";
+import { Dataset } from "../types";
+
+/**
+ * Series item type shared by builders that work only with the current frame (no accumulation).
+ * Used by IndexDatasetsBuilder and CurrentCustomDatasetsBuilder.
+ */
+const emptyPaths = new Set<string>();
+
+/**
+ * Rebuilds the series map from a new config array, preserving existing dataset data for
+ * matching keys and updating chart.js styling properties.
+ */
+export function setSeries(
+  existing: ReadonlyMap<SeriesConfigKey, CurrentFrameSeriesItem>,
+  series: Immutable<SeriesItem[]>,
+): Map<SeriesConfigKey, CurrentFrameSeriesItem> {
+  const newSeries = new Map<SeriesConfigKey, CurrentFrameSeriesItem>();
+
+  for (const item of series) {
+    let existingSeries = existing.get(item.key);
+    existingSeries ??= {
+      configIndex: item.configIndex,
+      enabled: item.enabled,
+      messagePath: item.messagePath,
+      parsed: item.parsed,
+      dataset: { data: [] },
+    };
+
+    existingSeries.configIndex = item.configIndex;
+    existingSeries.enabled = item.enabled;
+    existingSeries.dataset = {
+      ...existingSeries.dataset,
+      borderColor: item.color,
+      showLine: item.showLine,
+      fill: false,
+      borderWidth: item.lineSize,
+      pointRadius: item.lineSize * 1.2,
+      pointHoverRadius: 3,
+      pointBackgroundColor: item.showLine ? item.contrastColor : item.color,
+      pointBorderColor: "transparent",
+    };
+
+    newSeries.set(item.key, existingSeries);
+  }
+
+  return newSeries;
+}
+
+/**
+ * Builds a GetViewportDatasetsResult from a series map. No downsampling is performed —
+ * suitable for builders that process only the current frame.
+ */
+export function buildViewportDatasets(
+  seriesByKey: ReadonlyMap<SeriesConfigKey, CurrentFrameSeriesItem>,
+  pathsWithMismatchedDataLengths: ReadonlySet<string> = emptyPaths,
+): GetViewportDatasetsResult {
+  const datasets: Dataset[] = [];
+  for (const series of seriesByKey.values()) {
+    if (series.enabled) {
+      datasets[series.configIndex] = series.dataset as Dataset;
+    }
+  }
+  return { datasetsByConfigIndex: datasets, pathsWithMismatchedDataLengths };
+}
+
+export function lastMatchingTopic(
+  msgEvents: Immutable<MessageEvent[]>,
+  topic: string,
+): MessageEvent | undefined {
+  for (let i = msgEvents.length - 1; i >= 0; --i) {
+    const msgEvent = msgEvents[i]!;
+    if (msgEvent.topic === topic) {
+      return msgEvent;
+    }
+  }
+
+  return undefined;
+}
+
+type SeriesCurrentAction<T> =
+  | { type: "reset-current"; series: SeriesConfigKey }
+  | { type: "append-current"; series: SeriesConfigKey; items: T[] };
+
+/**
+ * Iterates over each series, optionally pushing a reset-current action when a seek occurred,
+ * then appending the items returned by `getItems`. Returns whether any series produced items.
+ *
+ * Shared by TimestampDatasetsBuilder and CustomDatasetsBuilder for current-frame y-series
+ * processing.
+ */
+export function buildCurrentSeriesActions<TItem>(
+  series: Immutable<Array<{ config: Immutable<SeriesItem> }>>,
+  options: { didSeek: boolean },
+  getItems: (config: Immutable<SeriesItem>) => TItem[],
+): { actions: SeriesCurrentAction<TItem>[]; datasetsChanged: boolean } {
+  let datasetsChanged = false;
+  const actions: SeriesCurrentAction<TItem>[] = [];
+  for (const s of series) {
+    if (options.didSeek) {
+      actions.push({ type: "reset-current", series: s.config.key });
+    }
+    const items = getItems(s.config);
+    datasetsChanged ||= items.length > 0;
+    actions.push({ type: "append-current", series: s.config.key, items });
+  }
+  return { actions, datasetsChanged };
+}
+
+type SeriesFullAction<T> =
+  | { type: "reset-full"; series: SeriesConfigKey }
+  | { type: "append-full"; series: SeriesConfigKey; items: T[] };
+
+/**
+ * Iterates over each series whose topic matches `topic`, optionally pushing a reset-full action
+ * when isReset is true, then appending non-empty items returned by `getItems`.
+ *
+ * Shared by TimestampDatasetsBuilder and CustomDatasetsBuilder for range y-series processing.
+ */
+export function buildFullSeriesActions<TItem>(
+  series: Immutable<Array<{ config: Immutable<SeriesItem> }>>,
+  topic: string,
+  options: { isReset: boolean },
+  getItems: (config: Immutable<SeriesItem>) => TItem[],
+): SeriesFullAction<TItem>[] {
+  const actions: SeriesFullAction<TItem>[] = [];
+  for (const s of series) {
+    if (s.config.parsed.topicName !== topic) {
+      continue;
+    }
+    if (options.isReset) {
+      actions.push({ type: "reset-full", series: s.config.key });
+    }
+    const items = getItems(s.config);
+    if (items.length > 0) {
+      actions.push({ type: "append-full", series: s.config.key, items });
+    }
+  }
+  return actions;
+}
diff --git a/packages/suite-base/src/panels/Plot/hooks/usePlotInteractionHandlers.test.ts b/packages/suite-base/src/panels/Plot/hooks/usePlotInteractionHandlers.test.ts
index be50d7e479..a6497f5863 100644
--- a/packages/suite-base/src/panels/Plot/hooks/usePlotInteractionHandlers.test.ts
+++ b/packages/suite-base/src/panels/Plot/hooks/usePlotInteractionHandlers.test.ts
@@ -451,7 +451,7 @@ describe("usePlotInteractionHandlers", () => {
       const { result, props } = setup({ coordinator: mockCoordinator });
 
       act(() => {
-        result.current.keyUphandlers.v();
+        result.current.keyUpHandlers.v();
       });
 
       expect(props.coordinator?.setZoomMode).toHaveBeenCalledWith("x");
@@ -461,7 +461,7 @@ describe("usePlotInteractionHandlers", () => {
       const { result, props } = setup({ coordinator: mockCoordinator });
 
       act(() => {
-        result.current.keyUphandlers.b();
+        result.current.keyUpHandlers.b();
       });
 
       expect(props.coordinator?.setZoomMode).toHaveBeenCalledWith("x");
diff --git a/packages/suite-base/src/panels/Plot/hooks/usePlotInteractionHandlers.ts b/packages/suite-base/src/panels/Plot/hooks/usePlotInteractionHandlers.ts
index 515a3673e5..83f818c371 100644
--- a/packages/suite-base/src/panels/Plot/hooks/usePlotInteractionHandlers.ts
+++ b/packages/suite-base/src/panels/Plot/hooks/usePlotInteractionHandlers.ts
@@ -188,7 +188,7 @@ const usePlotInteractionHandlers = ({
     setFocusedPath(["paths", String(index)]);
   }, []);
 
-  const { keyDownHandlers, keyUphandlers } = useMemo(() => {
+  const { keyDownHandlers, keyUpHandlers } = useMemo(() => {
     return {
       keyDownHandlers: {
         v: () => {
@@ -198,7 +198,7 @@ const usePlotInteractionHandlers = ({
           coordinator?.setZoomMode("xy");
         },
       },
-      keyUphandlers: {
+      keyUpHandlers: {
         v: () => {
           coordinator?.setZoomMode("x");
         },
@@ -244,7 +244,7 @@ const usePlotInteractionHandlers = ({
     onClickPath,
     focusedPath,
     keyDownHandlers,
-    keyUphandlers,
+    keyUpHandlers,
     getPanelContextMenuItems,
   };
 };
diff --git a/packages/suite-base/src/panels/Plot/types.ts b/packages/suite-base/src/panels/Plot/types.ts
index 06bfc7ccdb..fa60033423 100644
--- a/packages/suite-base/src/panels/Plot/types.ts
+++ b/packages/suite-base/src/panels/Plot/types.ts
@@ -129,7 +129,7 @@ export type UseHoverHandlersHook = {
     v: () => void;
     b: () => void;
   };
-  keyUphandlers: {
+  keyUpHandlers: {
     v: () => void;
     b: () => void;
   };