fix(iobroker): give the SDK sole reconnect ownership (#119)

StreamHandler ran its own five-attempt reconnect timer alongside the
SDK's unbounded exponential backoff, and re-registered every listener
on each scheduled reconnect (duplicate handlers over time). Listeners
now register exactly once in connect(); stream_error and terminal
auth_failure are handled so info.connection accurately reflects
transient-down vs permanently-stopped state.

Author: Bre77

.changeset/iob-reconnect-ownership.md

@@ -0,0 +1,5 @@
+---
+"iobroker.teslemetry": patch
+---
+
+Give the SDK sole ownership of stream reconnection instead of running a competing five-attempt reconnect timer alongside it: `StreamHandler` now registers its stream listeners exactly once (fixing double-registration on repeated `connect()` calls) and handles `stream_error` and terminal `auth_failure`, so `info.connection` accurately reflects whether the stream is transiently down or has stopped for good.

packages/iobroker.teslemetry/lib/StreamHandler.ts

@@ -3,10 +3,12 @@ import { StateManager } from './StateManager.js';
 import { EnergyHandler } from './EnergyHandler.js';
 
 export class StreamHandler {
-	private reconnectAttempts = 0;
-	private maxReconnectAttempts = 5;
-	private reconnectTimer?: NodeJS.Timeout;
-	private isConnecting = false;
+	// The SDK's TeslemetryStream owns reconnection (unbounded exponential
+	// backoff, then a permanent stop after two consecutive auth failures) -
+	// this class only listens and reflects that state, it never schedules
+	// its own reconnect. Listeners are registered once; sse.on() replays
+	// cached values into the handler and re-listening would double-dispatch.
+	private listenersRegistered = false;
 
 	constructor(
 		private adapter: ioBroker.Adapter,
@@ -16,67 +18,74 @@ export class StreamHandler {
 	) {}
 
 	/**
-	 * Connect to SSE stream and set up event handlers
+	 * Register stream event handlers (once) and connect to the SSE stream.
+	 * Safe to call again after a connect: the SDK's connect() is a no-op
+	 * while already active.
 	 */
 	async connect(): Promise<void> {
-		if (this.isConnecting) {
-			this.adapter.log.debug('Stream connection already in progress');
-			return;
+		const sse = this.teslemetry.sse;
+
+		if (!this.listenersRegistered) {
+			this.listenersRegistered = true;
+			this.registerListeners(sse);
 		}
 
-		this.isConnecting = true;
 		this.adapter.log.info('Connecting to Teslemetry SSE stream...');
+		await sse.connect();
+	}
 
-		try {
-			const sse = this.teslemetry.sse;
-
-			// Set up event handlers
-			sse.on('connect', () => {
-				this.adapter.log.info('SSE stream connected');
-				this.reconnectAttempts = 0;
-				this.isConnecting = false;
-				this.adapter.setStateAsync('info.connection', true, true);
-			});
-
-			sse.on('disconnect', () => {
-				this.adapter.log.warn('SSE stream disconnected');
-				this.isConnecting = false;
-				this.adapter.setStateAsync('info.connection', false, true);
-				this.scheduleReconnect();
-			});
-
-			// Handle vehicle data updates
-			sse.on('data', (event: any) => {
-				this.handleDataEvent(event);
-			});
-
-			// Handle vehicle state changes (online/asleep/offline)
-			sse.on('state', (event: any) => {
-				this.handleStateEvent(event);
-			});
-
-			// Handle alerts
-			sse.on('alerts', (event: any) => {
-				this.handleAlertEvent(event);
-			});
-
-			// Handle energy site live power/battery/grid updates
-			sse.on('live_status', (event: any) => {
-				this.handleLiveStatusEvent(event);
-			});
-
-			// Handle energy site settings updates (operation mode, reserves, ...)
-			sse.on('site_info', (event: any) => {
-				this.handleSiteInfoEvent(event);
-			});
-
-			// Connect to stream
-			await sse.connect();
-		} catch (error: any) {
-			this.adapter.log.error(`Failed to connect to SSE stream: ${error.message}`);
-			this.isConnecting = false;
-			this.scheduleReconnect();
-		}
+	private registerListeners(sse: Teslemetry['sse']): void {
+		sse.on('connect', () => {
+			this.adapter.log.info('SSE stream connected');
+			this.adapter.setStateAsync('info.connection', true, true);
+		});
+
+		// The SDK retries transient disconnects on its own; this only reflects
+		// current connectivity; it does not imply a terminal stop.
+		sse.on('disconnect', () => {
+			this.adapter.log.warn('SSE stream disconnected');
+			this.adapter.setStateAsync('info.connection', false, true);
+		});
+
+		sse.on('stream_error', ({ error, status, retries }) => {
+			const message = error instanceof Error ? error.message : String(error);
+			this.adapter.log.warn(
+				`SSE stream error (status ${status ?? 'unknown'}, attempt ${retries}): ${message}`
+			);
+		});
+
+		sse.on('auth_failure', (error) => {
+			this.adapter.log.error(
+				`SSE stream authentication failed twice in a row and has stopped permanently: ${error.message}. ` +
+					'Fix the access token and restart the adapter to resume streaming.'
+			);
+			this.adapter.setStateAsync('info.connection', false, true);
+		});
+
+		// Handle vehicle data updates
+		sse.on('data', (event: any) => {
+			this.handleDataEvent(event);
+		});
+
+		// Handle vehicle state changes (online/asleep/offline)
+		sse.on('state', (event: any) => {
+			this.handleStateEvent(event);
+		});
+
+		// Handle alerts
+		sse.on('alerts', (event: any) => {
+			this.handleAlertEvent(event);
+		});
+
+		// Handle energy site live power/battery/grid updates
+		sse.on('live_status', (event: any) => {
+			this.handleLiveStatusEvent(event);
+		});
+
+		// Handle energy site settings updates (operation mode, reserves, ...)
+		sse.on('site_info', (event: any) => {
+			this.handleSiteInfoEvent(event);
+		});
 	}
 
 	/**
@@ -85,11 +94,6 @@ export class StreamHandler {
 	disconnect(): void {
 		this.adapter.log.info('Disconnecting from SSE stream...');
 
-		if (this.reconnectTimer) {
-			clearTimeout(this.reconnectTimer);
-			this.reconnectTimer = undefined;
-		}
-
 		try {
 			this.teslemetry.sse.disconnect();
 			this.adapter.setStateAsync('info.connection', false, true);
@@ -202,38 +206,4 @@ export class StreamHandler {
 			this.adapter.log.error(`Error handling site_info event: ${error.message}`);
 		}
 	}
-
-	/**
-	 * Schedule automatic reconnection with exponential backoff
-	 */
-	private scheduleReconnect(): void {
-		if (this.reconnectAttempts >= this.maxReconnectAttempts) {
-			this.adapter.log.error(
-				`Max reconnection attempts (${this.maxReconnectAttempts}) reached. Stopping reconnection attempts.`
-			);
-			return;
-		}
-
-		if (this.reconnectTimer) {
-			clearTimeout(this.reconnectTimer);
-		}
-
-		const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
-		this.reconnectAttempts++;
-
-		this.adapter.log.info(
-			`Scheduling reconnection attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts} in ${delay / 1000}s`
-		);
-
-		this.reconnectTimer = setTimeout(() => {
-			this.connect();
-		}, delay);
-	}
-
-	/**
-	 * Reset reconnection attempts (call this after successful manual reconnection)
-	 */
-	resetReconnectAttempts(): void {
-		this.reconnectAttempts = 0;
-	}
 }

packages/iobroker.teslemetry/src/main.ts

@@ -153,13 +153,11 @@ class TeslemetryAdapter extends utils.Adapter {
 				this.pollInterval = undefined;
 			}
 
-			// Disconnect streaming
+			// Disconnect streaming (StreamHandler.disconnect() also closes the
+			// underlying SSE connection - no separate teslemetry.sse.disconnect() needed)
 			if (this.streamHandler) {
 				this.streamHandler.disconnect();
-			}
-
-			// Disconnect SSE
-			if (this.teslemetry) {
+			} else if (this.teslemetry) {
 				this.teslemetry.sse.disconnect();
 			}
 

packages/iobroker.teslemetry/test/StreamHandler.test.ts

@@ -127,3 +127,73 @@ test('live_status events for an unselected/unregistered energy site are ignored
 
 	assert.equal(states.get(`energy.${SITE_ID}.live.solar_power`), undefined);
 });
+
+const STREAM_EVENTS = ['connect', 'disconnect', 'stream_error', 'auth_failure', 'data', 'state', 'alerts', 'live_status', 'site_info'] as const;
+
+test('repeated disconnect/reconnect cycles keep listener counts constant (regression: connect() used to re-register a fresh set of listeners every time)', async () => {
+	const teslemetry = new Teslemetry('fake-token');
+	// The real SDK connect() no-ops while already active; simulate the SDK
+	// marking itself active on each call, as it would after a real reconnect.
+	(teslemetry.sse as any).connect = () => {
+		(teslemetry.sse as any).active = true;
+		return Promise.resolve();
+	};
+
+	const { adapter } = createFakeAdapter();
+	const stateManager = new StateManager(adapter);
+	const streamHandler = new StreamHandler(adapter, teslemetry, stateManager, new EnergyHandler(adapter, teslemetry, stateManager));
+
+	await streamHandler.connect();
+	// data/state/alerts/live_status/site_info also carry the SDK's own
+	// internal cache listener (TeslemetryStream's constructor), so their
+	// baseline is 2 (SDK + ours) rather than 1 - what matters is that a
+	// reconnect cycle never adds another one on top of that baseline.
+	const countsAfterFirstConnect = STREAM_EVENTS.map((event) => teslemetry.sse.listenerCount(event));
+	assert.ok(
+		countsAfterFirstConnect.every((count) => count >= 1),
+		`expected at least one listener per event after the first connect, got: ${JSON.stringify(countsAfterFirstConnect)}`,
+	);
+
+	for (let i = 0; i < 3; i++) {
+		teslemetry.sse.emit('disconnect');
+		await streamHandler.connect();
+	}
+
+	const countsAfterCycles = STREAM_EVENTS.map((event) => teslemetry.sse.listenerCount(event));
+	assert.deepEqual(countsAfterCycles, countsAfterFirstConnect);
+});
+
+test('stream_error is logged without scheduling a competing reconnect (regression: adapter used to run its own reconnect timer alongside the SDK)', async () => {
+	const teslemetry = new Teslemetry('fake-token');
+	(teslemetry.sse as any).connect = () => Promise.resolve();
+
+	const { adapter, logs } = createFakeAdapter();
+	const stateManager = new StateManager(adapter);
+	const streamHandler = new StreamHandler(adapter, teslemetry, stateManager, new EnergyHandler(adapter, teslemetry, stateManager));
+	await streamHandler.connect();
+
+	teslemetry.sse.emit('stream_error', { error: new Error('boom'), status: 503, retries: 2 } as any);
+	await new Promise((resolve) => setTimeout(resolve, 0));
+
+	assert.ok(logs.some((l) => l.level === 'warn' && l.message.includes('boom') && l.message.includes('503')));
+});
+
+test('terminal auth_failure marks info.connection false and logs an error (regression: adapter never listened for auth_failure at all)', async () => {
+	const teslemetry = new Teslemetry('fake-token');
+	(teslemetry.sse as any).connect = () => Promise.resolve();
+
+	const { adapter, states, logs } = createFakeAdapter();
+	const stateManager = new StateManager(adapter);
+	const streamHandler = new StreamHandler(adapter, teslemetry, stateManager, new EnergyHandler(adapter, teslemetry, stateManager));
+	await streamHandler.connect();
+
+	teslemetry.sse.emit('connect');
+	await new Promise((resolve) => setTimeout(resolve, 0));
+	assert.equal(states.get('info.connection'), true);
+
+	teslemetry.sse.emit('auth_failure', new Error('token expired') as any);
+	await new Promise((resolve) => setTimeout(resolve, 0));
+
+	assert.equal(states.get('info.connection'), false);
+	assert.ok(logs.some((l) => l.level === 'error' && l.message.includes('stopped permanently')));
+});