Last-Event-ID for Stream Resumption — Complete Guide
In this tutorial, you will learn about Last. We cover key concepts, practical examples, and best practices to help you master this topic.
Use Last-Event-ID for SSE stream resumption: track event IDs, replay missed events after reconnection, implement event stores, and build reliable streams that survive network interruptions.
What You Learn
You will learn how the Last-Event-ID header works, how to implement event stores for replay, how to resume streams after reconnection, handle stream gaps, and build reliable SSE connections.
Why It Matters
Without Last-Event-ID, reconnecting clients miss events sent during disconnection. For critical streams (alerts, notifications, trades), missed events mean data loss. Last-Event-ID enables reliable delivery.
Real-World Use
DodaTech's alert SSE stream tracks event IDs monotonically. When a dashboard client reconnects, it sends Last-Event-ID: 145. The server replays events 146 through the latest. The user sees no gaps in the alert timeline.
Server-Side ID Generation
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
import threading
class EventStore:
def __init__(self, max_events=1000):
self.events = []
self.max_events = max_events
self.counter = 0
self._lock = threading.Lock()
def add_event(self, event_type, data):
with self._lock:
self.counter += 1
entry = {
'id': self.counter,
'type': event_type,
'data': data,
'timestamp': time.time(),
}
self.events.append(entry)
if len(self.events) > self.max_events:
self.events.pop(0)
return entry
def get_events_since(self, last_id):
with self._lock:
start_id = int(last_id) if last_id else 0
return [e for e in self.events if e['id'] > start_id]
event_store = EventStore()
class LastEventIDSSE(BaseHTTPRequestHandler):
def do_GET(self):
last_event_id = self.headers.get('Last-Event-ID')
print(f"Client request, Last-Event-ID: {last_event_id}")
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
self.end_headers()
# Replay missed events
missed = event_store.get_events_since(last_event_id)
for event in missed:
self._send_event(event['id'], event['type'], event['data'])
# Stream new events
last_sent_id = missed[-1]['id'] if missed else (int(last_event_id) if last_event_id else 0)
while True:
new_events = event_store.get_events_since(last_sent_id)
for event in new_events:
self._send_event(event['id'], event['type'], event['data'])
last_sent_id = event['id']
time.sleep(0.5)
def _send_event(self, event_id, event_type, data):
self.wfile.write(f"id: {event_id}\n".encode())
self.wfile.write(f"event: {event_type}\n".encode())
self.wfile.write(f"data: {json.dumps(data)}\n\n".encode())
Event Replay with Bounded Store
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
from collections import deque
class BoundedEventStore:
def __init__(self, max_size=500):
self.events = deque(maxlen=max_size)
self.counter = 0
def append(self, event_type, data):
self.counter += 1
self.events.append({
'id': self.counter,
'type': event_type,
'data': data,
'time': time.time(),
})
return self.counter
def since(self, last_id):
if not last_id:
return list(self.events)
start = int(last_id)
return [e for e in self.events if e['id'] > start]
def get_last_id(self):
if self.events:
return self.events[-1]['id']
return 0
store = BoundedEventStore(max_size=100)
# Simulate events being added from other threads
def simulate_events():
for i in range(20):
store.append('metric', {'value': i * 10, 'time': time.time()})
time.sleep(0.3)
threading.Thread(target=simulate_events, daemon=True).start()
class ReplaySSE(BaseHTTPRequestHandler):
def do_GET(self):
last_id = self.headers.get('Last-Event-ID')
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.end_headers()
missed = store.since(last_id)
for event in missed:
self.wfile.write(f"id: {event['id']}\nevent: {event['type']}\ndata: {json.dumps(event['data'])}\n\n".encode())
time.sleep(0.1)
last_sent = missed[-1]['id'] if missed else store.get_last_id()
while True:
new_events = store.since(last_sent)
for event in new_events:
self.wfile.write(f"id: {event['id']}\nevent: {event['type']}\ndata: {json.dumps(event['data'])}\n\n".encode())
last_sent = event['id']
time.sleep(0.5)
Client-Side Resilience
class ResilientEventSource {
constructor(url, options = {}) {
this.url = url;
this.options = options;
this.lastEventId = null;
this.eventLog = [];
this.connect();
}
connect() {
// Pass last event ID as URL parameter since
// the browser handles Last-Event-ID automatically
const url = this.lastEventId
? `${this.url}?lastId=${this.lastEventId}`
: this.url;
this.source = new EventSource(url);
this.source.onmessage = (event) => {
if (event.lastEventId) {
this.lastEventId = event.lastEventId;
}
this.eventLog.push({
id: event.lastEventId,
data: event.data,
time: Date.now(),
});
if (this.eventLog.length > 100) {
this.eventLog.shift();
}
if (this.options.onMessage) {
this.options.onMessage(event);
}
};
this.source.onerror = () => {
if (this.options.onReconnecting) {
this.options.onReconnecting(this.lastEventId);
}
};
this.source.onopen = () => {
if (this.options.onReconnected && this.lastEventId) {
this.options.onReconnected(this.lastEventId);
}
};
}
getMissedEvents() {
return this.eventLog.slice(-50);
}
close() {
if (this.source) {
this.source.close();
}
}
}
const events = new ResilientEventSource('/stream', {
onReconnecting: (lastId) => {
console.log(`Reconnecting from event ${lastId}`);
showStatus('reconnecting');
},
onReconnected: (lastId) => {
console.log(`Reconnected, resumed from ${lastId}`);
showStatus('connected');
},
onMessage: (event) => {
updateUI(JSON.parse(event.data));
},
});
Database-Backed Event Store
import sqlite3
import json
import time
class PersistentEventStore:
def __init__(self, db_path='/tmp/sse_events.db'):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.execute('''
CREATE TABLE IF NOT EXISTS sse_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
event_type TEXT NOT NULL,
data TEXT NOT NULL,
created_at REAL NOT NULL
)
''')
self.conn.execute('''
CREATE INDEX IF NOT EXISTS idx_events_id ON sse_events(id)
''')
self.conn.commit()
self._lock = threading.Lock()
def add_event(self, event_type, data):
with self._lock:
cursor = self.conn.execute(
'INSERT INTO sse_events (event_type, data, created_at) VALUES (?, ?, ?)',
(event_type, json.dumps(data), time.time())
)
self.conn.commit()
return cursor.lastrowid
def get_events_since(self, last_id):
with self._lock:
start_id = int(last_id) if last_id else 0
cursor = self.conn.execute(
'SELECT id, event_type, data, created_at FROM sse_events WHERE id > ? ORDER BY id',
(start_id,)
)
return [
{'id': row[0], 'type': row[1], 'data': json.loads(row[2]), 'time': row[3]}
for row in cursor.fetchall()
]
def cleanup(self, max_age_hours=24):
cutoff = time.time() - (max_age_hours * 3600)
self.conn.execute('DELETE FROM sse_events WHERE created_at < ?', (cutoff,))
self.conn.commit()
store = PersistentEventStore()
store.add_event('notification', {'message': 'Server started'})
store.add_event('metric', {'cpu': 45.2, 'memory': 62.1})
for event in store.get_events_since(0):
print(f"Event {event['id']}: {event['type']} -> {event['data']}")
Expected output:
Event 1: notification -> {'message': 'Server started'}
Event 2: metric -> {'cpu': 45.2, 'memory': 62.1}
Common Mistakes
1. Non-Monotonic Event IDs
Event IDs must increase monotonically. Using timestamps with millisecond precision can create duplicate or out-of-order IDs.
2. Unlimited Event Storage
Storing all events forever consumes memory. Use a bounded store with a maximum size or TTL.
3. Not Handling ID Overflow
Event IDs stored as integers will eventually overflow. Use a large integer type or reset periodically with client notification.
4. Replaying Without Context
Replaying events without considering their sequence can cause incorrect UI state. Design events to be idempotent.
5. Ignoring Last-Event-ID on First Connection
On first connection, there is no Last-Event-ID. The server should not replay old events. Only replay when lastId > 0.
Practice Questions
1. How does the browser send Last-Event-ID on reconnection?
It automatically includes the Last-Event-ID HTTP header with the value of the last received event's id field. The server reads this header.
2. Why use a bounded event store?
To limit memory usage. Old events are no longer needed because clients have either received them or reconnected and caught up.
3. What makes a good event ID scheme?
Monotonically increasing, gap-free, and unique. Integers work well. UUIDs are not suitable because they do not indicate order.
4. How do you handle event ID overflow?
Use 64-bit integers (they will not overflow in practice). Or reset the counter and notify clients to discard their cached last ID.
Challenge
Build a persistent event store for SSE with: SQLite backend, 50,000 event limit, cleanup of events older than 7 days, replay from any Last-Event-ID, batch insertion of events, and a status endpoint showing store size and latest ID.
FAQ
Mini Project: Reliable SSE Stream
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
import threading
from collections import deque
class ReliableSSE:
def __init__(self, max_events=500):
self.store = deque(maxlen=max_events)
self.counter = 0
self._lock = threading.Lock()
def add(self, event_type, data):
with self._lock:
self.counter += 1
event = {'id': self.counter, 'type': event_type, 'data': data, 'time': time.time()}
self.store.append(event)
return event
def since(self, last_id):
with self._lock:
if not last_id:
return []
start = int(last_id)
return [e for e in self.store if e['id'] > start]
sse_store = ReliableSSE()
class ReliableSSEHandler(BaseHTTPRequestHandler):
def do_GET(self):
last_id = self.headers.get('Last-Event-ID')
self.send_response(200)
self.send_header('Content-Type', 'text/event-stream')
self.send_header('Cache-Control', 'no-cache')
self.send_header('Connection', 'keep-alive')
self.end_headers()
missed = sse_store.since(last_id)
for event in missed:
self.wfile.write(f"id: {event['id']}\nevent: {event['type']}\ndata: {json.dumps(event['data'])}\n\n".encode())
time.sleep(0.05)
last_sent = missed[-1]['id'] if missed else (int(last_id) if last_id else sse_store.counter)
while True:
new_events = sse_store.since(last_sent)
for event in new_events:
self.wfile.write(f"id: {event['id']}\nevent: {event['type']}\ndata: {json.dumps(event['data'])}\n\n".encode())
last_sent = event['id']
time.sleep(0.5)
threading.Thread(target=lambda: [sse_store.add('ping', {'t': time.time()}) or time.sleep(2) for _ in iter(int, 1)], daemon=True).start()
server = HTTPServer(('localhost', 3000), ReliableSSEHandler)
print("Reliable SSE on :3000")
server.serve_forever()
What's Next
Now that you understand Last-Event-ID, explore SSE headers and configuration, then learn about SSE and CORS.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro