Syncing SSE Events into Pinia Stores

Part of State-Management Integration for SSE, under Frontend Consumption & Client Patterns.

Pinia stores outlive components, which makes them the right home for a stream — and the wrong place to open one carelessly, because a connection created in a store lives until the page unloads whether anyone is watching or not.

Symptom & Developer Intent

  • Navigating between routes opens a new stream each time and never closes the old one.
  • Streamed data updates in the store but the view does not re-render.
  • Vue devtools shows thousands of reactive proxies after a busy session.
  • Reconnects replay events and counters double-count.
  • Two components each start the stream, so every event is applied twice.

The intent is one connection owned by the store layer, reactive updates that are cheap, and idempotent application of replayed events.

Root Cause Analysis

Three distinct causes sit behind those symptoms.

Three symptoms, three unrelated causes Table pairing the connection leak, the non-updating view and the double-counting bug with their causes and fixes in a Pinia integration. Three symptoms, three unrelated causes Cause Fix Stream per route change store has no lifecycle reference counting View stops updating reactive reassigned mutate + triggerRef Counters double replay applied twice dedupe by event id Thousands of proxies deep reactivity shallowRef
They arrive together in bug reports and share no root cause, which is why fixing one never seems to help with the others.

Ownership. A component that starts a stream can close it on unmount; a store cannot, because it has no lifecycle. Without an explicit reference count, the connection either leaks on every route change or is closed by the first component to leave while others are still watching.

Reactivity cost. Pinia state is deeply reactive by default, so every streamed payload is wrapped in a proxy on assignment — recursively. For append-only stream data nothing reads those inner keys reactively, so the proxy work is pure overhead, and at a few hundred events a second it is visible.

Idempotence. Reconnect replay is inclusive at the boundary: the server resends the event matching the cursor. A reducer that increments rather than sets will double-count it, and the drift is permanent.

A fourth, subtler issue is reassignment. Replacing a reactive() object breaks the binding that components hold, so the view silently stops updating — the Vue-specific version of “the data is there but nothing renders”.

Step-by-Step Resolution

Reference counting the shared connection State progression of a shared stream as components mount and unmount, opening on the first subscriber and closing on the last release. Reference counting the shared connection Idle 0 subscribers Opened first connect() component mounts Shared n subscribers, 1 stream more mount Closed last release() all unmounted the next mount opens a fresh stream from the stored cursor
A store cannot close what it does not know is unused. The count is the missing lifecycle, and it is four lines.

Step 1 — Put the connection in its own store with reference counting 

// stores/stream.js
import { defineStore } from 'pinia';
import { ref, shallowRef } from 'vue';

export const useStreamStore = defineStore('stream', () => {
  const status = ref('idle');            // idle | connecting | live | retrying | offline
  const lastEventId = ref(null);
  const es = shallowRef(null);           // shallowRef: never proxy the EventSource itself
  let subscribers = 0;

  function connect() {
    subscribers += 1;
    if (es.value) return release;        // already connected: just count the subscriber

    status.value = 'connecting';
    const url = new URL('/api/stream', location.origin);
    if (lastEventId.value) url.searchParams.set('lastEventId', lastEventId.value);

    const source = new EventSource(url, { withCredentials: true });

    source.addEventListener('open', () => { status.value = 'live'; });
    source.addEventListener('error', () => {
      status.value = source.readyState === EventSource.CLOSED ? 'offline' : 'retrying';
    });
    source.addEventListener('resync', () => { useOrdersStore().resync(); });

    // Route domain events to the stores that own them.
    source.addEventListener('order', (e) => {
      lastEventId.value = e.lastEventId || lastEventId.value;
      useOrdersStore().applyEvent(JSON.parse(e.data), e.lastEventId);
    });

    es.value = source;
    return release;
  }

  function release() {
    subscribers = Math.max(0, subscribers - 1);
    if (subscribers === 0) {             // last component out closes the stream
      es.value?.close();
      es.value = null;
      status.value = 'idle';
    }
  }

  return { status, lastEventId, connect, release };
});

Step 2 — Connect and release from the component lifecycle 

<script setup>
import { onMounted, onUnmounted } from 'vue';
import { useStreamStore } from '@/stores/stream';

const stream = useStreamStore();
let release;

onMounted(() => { release = stream.connect(); });
onUnmounted(() => { release?.(); });      // reference count, not an unconditional close
</script>

Step 3 — Keep streamed payloads out of deep reactivity 

// stores/orders.js
import { defineStore } from 'pinia';
import { shallowRef, triggerRef } from 'vue';

export const useOrdersStore = defineStore('orders', () => {
  // A Map in a shallowRef: no per-payload proxy, one explicit trigger per batch.
  const byId = shallowRef(new Map());
  const seen = new Set();                 // event ids already applied

  function applyEvent(order, eventId) {
    if (eventId && seen.has(eventId)) return;   // replayed event: drop it
    if (eventId) {
      seen.add(eventId);
      if (seen.size > 5000) seen.delete(seen.values().next().value);   // bounded
    }
    byId.value.set(order.id, order);      // mutate the Map…
    triggerRef(byId);                     // …then tell Vue once
  }

  async function resync() {
    const snapshot = await (await fetch('/api/orders')).json();
    byId.value = new Map(snapshot.map((o) => [o.id, o]));
    seen.clear();
    triggerRef(byId);
  }

  return { byId, applyEvent, resync };
});

Step 4 — Batch bursts onto the animation frame 

// Coalesce a burst into one trigger so the view renders at most once per frame.
let pending = [];
let queued = false;

function applyEventBatched(order, eventId) {
  pending.push([order, eventId]);
  if (queued) return;
  queued = true;
  requestAnimationFrame(() => {
    queued = false;
    const batch = pending;
    pending = [];
    for (const [o, id] of batch) applyEventNoTrigger(o, id);
    triggerRef(byId);                     // one trigger for the whole batch
  });
}

Step 5 — Read through computed properties, not by reassignment 

<script setup>
import { computed } from 'vue';
import { useOrdersStore } from '@/stores/orders';

const orders = useOrdersStore();
// Derived views stay reactive because they read through the ref, never replace it.
const openOrders = computed(() =>
  [...orders.byId.values()].filter((o) => o.status === 'open'),
);
</script>

Validation & Monitoring

Cost of storing one streamed event Bar chart comparing the relative cost of storing a 30-key event payload in a deep reactive object, a ref and a shallowRef with an explicit trigger. Cost of storing one streamed event one 30-key JSON payload, relative cost reactive (deep) 12× ref 3× shallowRef + triggerRef 1× relative cost per stored event
Deep reactivity walks every key of every payload to build proxies that nothing reads reactively — pure overhead on append-only stream data.
// 1. Reference counting: mounting two consumers must open exactly one connection.
const a = stream.connect();
const b = stream.connect();
console.assert(countOpenEventSources() === 1, 'more than one connection opened');
a(); console.assert(countOpenEventSources() === 1, 'closed while still in use');
b(); console.assert(countOpenEventSources() === 0, 'not closed by the last release');

// 2. Idempotence: applying the same event twice must not change the store twice.
orders.applyEvent({ id: 7, total: 100 }, 'evt-1');
orders.applyEvent({ id: 7, total: 100 }, 'evt-1');
console.assert(orders.byId.size === 1);

In Vue devtools, the timeline shows store mutations; a burst that produces one mutation rather than one per event confirms the batching works. The Components panel showing thousands of proxy objects is the signal that deep reactivity is still wrapping payloads.

Route changes are the practical regression test: navigate around the application for a minute and check that the Network panel shows one stream, not one per visited route.

Production Checklist

Frequently Asked Questions

Should the EventSource live in a store or in a composable?

In a store, with reference counting. A composable is scoped to a component, so several components mean several connections; a store gives one shared connection, but because a store has no lifecycle it needs an explicit count so the last consumer closes it.

Why did my view stop updating after I assigned new data?

Because you replaced a reactive container instead of mutating it. Components hold a reference to the original object, and reassigning leaves them bound to the old one. Mutate the Map or array inside the ref and call triggerRef, or replace ref.value rather than the object a component captured.

Is shallowRef really necessary for streamed data?

Once event rates reach the hundreds per second, yes. Deep reactivity walks every key of every payload to build proxies, and for append-only stream data nothing reads those inner keys reactively — so the work is entirely wasted. A shallow container plus an explicit trigger gives the same view updates at a fraction of the cost.

How large should the deduplication set be?

Large enough to cover a replay window, which is a few hundred to a few thousand ids for most feeds. Bound it — an unbounded set is a slow memory leak — and clear it on resync, since a snapshot makes every earlier id irrelevant.