From afcc98a95ac621911e0e2745e5e15c8b29c2d106 Mon Sep 17 00:00:00 2001 From: Wolf Mermelstein Date: Wed, 28 Oct 2026 11:38:20 +0000 Subject: [PATCH 1/5] fix(web): keep presence heartbeats while the tab is visible Why: the entity heartbeat paused whenever the window lost focus, so a viewer reading a document beside the window they were typing in (a second monitor, a call) dropped out of everyone's header avatars after 60s. About a third of heartbeat lapses in presence telemetry came from visible, unfocused tabs. What: gate the heartbeat and the catch-up ping on page visibility instead of window focus. --- apps/web/src/lib/core/signal/tabFocus.ts | 10 +++++ .../messages/tests/subscription.test.ts | 44 ++++++++++++------- .../service-connection/client.ts | 14 +++--- 3 files changed, 45 insertions(+), 23 deletions(-) diff --git a/apps/web/src/lib/core/signal/tabFocus.ts b/apps/web/src/lib/core/signal/tabFocus.ts index 7fd950d292e..f89a90a8cce 100644 --- a/apps/web/src/lib/core/signal/tabFocus.ts +++ b/apps/web/src/lib/core/signal/tabFocus.ts @@ -20,6 +20,16 @@ createEffect(() => { }); }); +const [isTabVisible_, setIsTabVisible] = createSignal( + document.visibilityState === 'visible' +); +document.addEventListener('visibilitychange', () => + setIsTabVisible(document.visibilityState === 'visible') +); + +/** Whether the tab is on screen, focused or not (e.g. beside a focused window) */ +export const isTabVisible = isTabVisible_; + /** Whether the tab is currently focused */ export const isTabFocused = createMemo(() => isTabFocused_(), { equalFn: (a: boolean | undefined, b: boolean | undefined) => a! === b!, diff --git a/apps/web/src/lib/queries/messages/tests/subscription.test.ts b/apps/web/src/lib/queries/messages/tests/subscription.test.ts index 5c663fa4597..700e5c6388f 100644 --- a/apps/web/src/lib/queries/messages/tests/subscription.test.ts +++ b/apps/web/src/lib/queries/messages/tests/subscription.test.ts @@ -10,12 +10,12 @@ import { useMessageThreadQuery } from '../thread-replies'; const mocks = await vi.hoisted(async () => { const { createSignal } = await import('solid-js'); - const [focused, setFocused] = createSignal(true); + const [visible, setVisible] = createSignal(true); return { send: vi.fn(), clearStream: vi.fn(), - focused, - setFocused, + visible, + setVisible, invalidate: vi.fn().mockResolvedValue(undefined), updates: new Set<(event: unknown) => void>(), reconnects: new Set<() => void>(), @@ -25,7 +25,11 @@ const mocks = await vi.hoisted(async () => { vi.mock('@macro-inc/observability', () => ({ Telemetry: { info: vi.fn(), warn: mocks.warn, error: vi.fn() }, })); -vi.mock('@core/signal/tabFocus', () => ({ isTabFocused: mocks.focused })); +// Focus is elsewhere throughout: presence follows visibility alone. +vi.mock('@core/signal/tabFocus', () => ({ + isTabFocused: () => false, + isTabVisible: mocks.visible, +})); vi.mock('@service-connection/stream', () => ({ clearStream: mocks.clearStream, })); @@ -85,7 +89,7 @@ function sent(action: string, id = parent.id) { beforeEach(() => { vi.useFakeTimers(); vi.clearAllMocks(); - mocks.setFocused(true); + mocks.setVisible(true); }); afterEach(() => { for (const dispose of disposers.splice(0)) dispose(); @@ -213,17 +217,23 @@ it('refreshes cached messages after all views close and the parent reopens', () expect(mocks.invalidate).toHaveBeenCalledTimes(3); }); +it('keeps the heartbeat in a visible tab whose window is not focused', () => { + mountBlock(); + vi.advanceTimersByTime(60_000); + expect(sent('ping')).toHaveLength(3); +}); + it('suspends the shared heartbeat in a background tab', () => { mountBlock(); - mocks.setFocused(false); + mocks.setVisible(false); vi.advanceTimersByTime(40_000); expect(sent('ping')).toHaveLength(0); - mocks.setFocused(true); + mocks.setVisible(true); vi.advanceTimersByTime(20_000); expect(sent('ping')).toHaveLength(1); }); -it('reports a presence lapse once per unfocused absence past the gateway window', () => { +it('reports a presence lapse once per hidden absence past the gateway window', () => { mount(useRefreshTrackedEntitiesOnFocus); mountBlock(); const lapses = () => @@ -231,7 +241,7 @@ it('reports a presence lapse once per unfocused absence past the gateway window' ([name]) => name === 'presence.heartbeat_lapsed' ); - mocks.setFocused(false); + mocks.setVisible(false); vi.advanceTimersByTime(40_000); expect(lapses()).toHaveLength(0); vi.advanceTimersByTime(40_000); @@ -243,13 +253,13 @@ it('reports a presence lapse once per unfocused absence past the gateway window' vi.advanceTimersByTime(60_000); expect(lapses()).toHaveLength(1); - mocks.setFocused(true); - mocks.setFocused(false); + mocks.setVisible(true); + mocks.setVisible(false); vi.advanceTimersByTime(60_000); expect(lapses()).toHaveLength(2); }); -it('pings on refocus and refreshes only views left unseen past the gateway window', () => { +it('pings when visible again and refreshes only views left unseen past the gateway window', () => { mount(useRefreshTrackedEntitiesOnFocus); mount(() => useMessageThreadQuery( @@ -259,20 +269,20 @@ it('pings on refocus and refreshes only views left unseen past the gateway windo ); mocks.invalidate.mockClear(); - // Heartbeats keep the gateway's view fresh while focused. + // Heartbeats keep the gateway's view fresh while visible, focused or not. vi.advanceTimersByTime(80_000); expect(sent('ping')).toHaveLength(4); - mocks.setFocused(false); + mocks.setVisible(false); vi.advanceTimersByTime(30_000); - mocks.setFocused(true); + mocks.setVisible(true); expect(sent('ping')).toHaveLength(5); expect(mocks.invalidate).not.toHaveBeenCalled(); // Away past the window: the gateway stopped sending this parent's events. - mocks.setFocused(false); + mocks.setVisible(false); vi.advanceTimersByTime(60_000); expect(sent('ping')).toHaveLength(5); - mocks.setFocused(true); + mocks.setVisible(true); expect(sent('ping')).toHaveLength(6); expect(mocks.invalidate).toHaveBeenCalledTimes(3); expect(mocks.invalidate).toHaveBeenCalledWith({ diff --git a/apps/web/src/lib/service-clients/service-connection/client.ts b/apps/web/src/lib/service-clients/service-connection/client.ts index a1223220aca..fee2724b670 100644 --- a/apps/web/src/lib/service-clients/service-connection/client.ts +++ b/apps/web/src/lib/service-clients/service-connection/client.ts @@ -1,4 +1,4 @@ -import { isTabFocused } from '@core/signal/tabFocus'; +import { isTabVisible } from '@core/signal/tabFocus'; import type { EntityId } from '@core/types'; import { fireAndForget } from '@core/util/fireAndForget'; import { createReconnectEffect } from '@macro-inc/collaboration/websocket'; @@ -51,8 +51,10 @@ export const connectionGatewayClient = { trackedEntities.set(args.entity_id, { entityType: args.entity_type, count: 1, + // Visible, not focused: someone reading beside the window they are + // typing in is still looking, and should keep their presence dot. heartbeat: setInterval(() => { - if (isTabFocused()) + if (isTabVisible()) fireAndForget( connectionGatewayClient.trackEntity({ ...args, @@ -171,16 +173,16 @@ export function useReopenTrackedEntitiesOnReconnect(): void { } /** - * Pings every tracked entity as soon as the tab regains focus, since the heartbeat pauses in + * Pings every tracked entity as soon as the tab is visible again, since the heartbeat pauses in * the background. A tab that went unseen longer than the gateway's activity window was left * out of the entity's events meanwhile, so its views refresh as they do after a reconnect. */ export function useRefreshTrackedEntitiesOnFocus(): void { createEffect( on( - isTabFocused, - (focused) => { - if (!focused) return; + isTabVisible, + (visible) => { + if (!visible) return; const now = Date.now(); for (const [entity_id, tracked] of trackedEntities) { const entity = { entity_id, entity_type: tracked.entityType }; From c436769a970c6f976c443d6f22b3288882c0d1e6 Mon Sep 17 00:00:00 2001 From: Wolf Mermelstein Date: Wed, 28 Oct 2026 11:38:21 +0000 Subject: [PATCH 2/5] test(web): cover live indicator store updates per entity Why: header avatars must re-render on every user_tracking_change for their entity; a store read that stops tracking would leave stale dots. What: add a store test that renders an entity's indicators through a series of tracking changes, including one for another entity. --- .../src/lib/core/state/liveIndicators.test.ts | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 apps/web/src/lib/core/state/liveIndicators.test.ts diff --git a/apps/web/src/lib/core/state/liveIndicators.test.ts b/apps/web/src/lib/core/state/liveIndicators.test.ts new file mode 100644 index 00000000000..a7e9655279b --- /dev/null +++ b/apps/web/src/lib/core/state/liveIndicators.test.ts @@ -0,0 +1,63 @@ +import { createRenderEffect, createRoot } from 'solid-js'; +import { expect, it, vi } from 'vitest'; + +const mocks = vi.hoisted(() => ({ + trackingChange: undefined as ((event: { data: string }) => void) | undefined, +})); +vi.mock('@core/constant/featureFlags', () => ({ + ENABLE_LIVE_INDICATORS: true, +})); +vi.mock('@service-connection/websocket', () => ({ ws: {} })); +vi.mock('@service-connection/presence-telemetry', () => ({ + reportTrackingChange: vi.fn(), + reportInvalidTrackingChange: vi.fn(), +})); +vi.mock('@macro-inc/collaboration/websocket', () => ({ + createWebsocketEventEffect: ( + _ws: unknown, + _type: string, + callback: (event: { data: string }) => void + ) => { + mocks.trackingChange = callback; + }, +})); + +const { useUserIndicators } = await import('./liveIndicators'); + +it('re-renders an entity viewers on each change to that entity only', () => { + const rendered: (string[] | undefined)[] = []; + const dispose = createRoot((dispose) => { + const indicators = useUserIndicators(() => 'document-id'); + createRenderEffect(() => rendered.push(indicators())); + return dispose; + }); + + mocks.trackingChange!({ + data: JSON.stringify({ + entity_id: 'document-id', + entity_type: 'document', + user_ids: ['macro|first@example.com'], + }), + }); + mocks.trackingChange!({ + data: JSON.stringify({ + entity_id: 'document-id', + entity_type: 'document', + user_ids: ['macro|first@example.com', 'macro|second@example.com'], + }), + }); + mocks.trackingChange!({ + data: JSON.stringify({ + entity_id: 'other-document-id', + entity_type: 'document', + user_ids: [], + }), + }); + dispose(); + + expect(rendered).toEqual([ + undefined, + ['macro|first@example.com'], + ['macro|first@example.com', 'macro|second@example.com'], + ]); +}); From b9cbb2a9bc5f85061a27fc433e2131980ab505a4 Mon Sep 17 00:00:00 2001 From: Wolf Mermelstein Date: Wed, 28 Oct 2026 11:52:45 +0000 Subject: [PATCH 3/5] fix(connection_gateway): count pings stamped by a faster clock as active Why: an entity row's age was computed with an unchecked u64 subtraction. Gateway instances stamp last_ping with their own clocks, so a fresh ping read on an instance whose clock lags wraps to a huge age in release builds and the viewer is left out of user_tracking_change. What: compute the age with saturating_sub through a pure is_active_at(now) and test the skew, expiry, and never-pinged cases. --- .../src/model/connection.rs | 24 ++++++--- .../src/model/connection/tests.rs | 51 +++++++++++++++++++ 2 files changed, 67 insertions(+), 8 deletions(-) create mode 100644 services/connection_gateway/src/model/connection/tests.rs diff --git a/services/connection_gateway/src/model/connection.rs b/services/connection_gateway/src/model/connection.rs index 863bb7beb1a..34b9ffbf245 100644 --- a/services/connection_gateway/src/model/connection.rs +++ b/services/connection_gateway/src/model/connection.rs @@ -39,15 +39,20 @@ impl StoredConnectionEntity { .duration_since(UNIX_EPOCH) .unwrap() .as_millis() as u64; + self.is_active_at(current_timestamp, threshold) + } - match (threshold, self.last_ping) { - // There is a threshold and the last ping is older than the threshold - (Some(t), Some(last_ping)) => (current_timestamp - last_ping) < t, - // There is no threshold compare the current timestamp with the created_at - (Some(t), None) => (current_timestamp - self.created_at) < t, - // No threshold then just return true - _ => true, - } + /// Whether this entry was opened or pinged within `threshold` of `now`. + /// + /// Gateway instances stamp `last_ping` with their own clocks, so a fresh + /// ping can read as slightly in the future on another instance; that + /// counts as active rather than wrapping into a huge age. + fn is_active_at(&self, now: u64, threshold: Option) -> bool { + let Some(threshold) = threshold else { + return true; + }; + let last_seen = self.last_ping.unwrap_or(self.created_at); + now.saturating_sub(last_seen) < threshold } } @@ -91,3 +96,6 @@ impl AsRef for ConnectionContext<'_> { self.config } } + +#[cfg(test)] +mod tests; diff --git a/services/connection_gateway/src/model/connection/tests.rs b/services/connection_gateway/src/model/connection/tests.rs new file mode 100644 index 00000000000..b29f97e09dd --- /dev/null +++ b/services/connection_gateway/src/model/connection/tests.rs @@ -0,0 +1,51 @@ +use super::*; + +#[test] +fn a_ping_stamped_ahead_of_this_instance_clock_is_active() { + let entry = StoredConnectionEntity { + pk: "#document#doc".to_owned(), + sk: "connection".to_owned(), + entity_type: EntityType::Document, + entity_id: "doc".to_owned(), + connection_id: "connection".to_owned(), + created_at: 1_000_000, + user_id: "macro|user@example.com".to_owned(), + last_ping: Some(1_000_500), + }; + + assert!(entry.is_active_at(1_000_000, Some(60_000))); +} + +#[test] +fn an_entry_not_pinged_within_the_threshold_is_inactive() { + let entry = StoredConnectionEntity { + pk: "#document#doc".to_owned(), + sk: "connection".to_owned(), + entity_type: EntityType::Document, + entity_id: "doc".to_owned(), + connection_id: "connection".to_owned(), + created_at: 1_000_000, + user_id: "macro|user@example.com".to_owned(), + last_ping: Some(1_000_000), + }; + + assert!(entry.is_active_at(1_059_999, Some(60_000))); + assert!(!entry.is_active_at(1_060_000, Some(60_000))); +} + +#[test] +fn an_open_entry_never_pinged_counts_from_creation() { + let entry = StoredConnectionEntity { + pk: "#document#doc".to_owned(), + sk: "connection".to_owned(), + entity_type: EntityType::Document, + entity_id: "doc".to_owned(), + connection_id: "connection".to_owned(), + created_at: 1_000_000, + user_id: "macro|user@example.com".to_owned(), + last_ping: None, + }; + + assert!(entry.is_active_at(999_000, Some(60_000))); + assert!(!entry.is_active_at(1_060_000, Some(60_000))); +} From 569816df79fbf7fdc53bf271429551d114c5c134 Mon Sep 17 00:00:00 2001 From: Wolf Mermelstein Date: Wed, 28 Oct 2026 11:52:49 +0000 Subject: [PATCH 4/5] fix(connection_gateway): re-create a missing presence row on ping Why: a ping for an entity row that no longer existed failed its condition silently, so a viewer whose open was lost (a swallowed write error, a row cleared under a live connection) stayed invisible until they reopened the entity. The entity list read right after an open or close was also eventually consistent and could miss that write. What: update_last_entity_ping returns None for a missing row and refresh_connection_entity writes the row back; get_entries_by_entity reads consistently. Document the presence rules in AGENTS.md. --- services/connection_gateway/AGENTS.md | 4 + .../src/service/connection.rs | 36 ++++-- .../src/service/connection/test.rs | 113 +++++++++++++++++- .../src/service/dynamodb.rs | 31 ++--- .../connection_gateway/src/service/tracker.rs | 2 +- 5 files changed, 161 insertions(+), 25 deletions(-) diff --git a/services/connection_gateway/AGENTS.md b/services/connection_gateway/AGENTS.md index 2975e6ae053..7ff9dc868eb 100644 --- a/services/connection_gateway/AGENTS.md +++ b/services/connection_gateway/AGENTS.md @@ -50,6 +50,10 @@ the socket themselves. - A send can target a connection on a different instance — it always goes through the Redis relay in `src/service/redis.rs`, never a direct in-process lookup; don't bypass it when adding a new send path. +- Presence (`user_tracking_change`) counts an entity row as active for 60s + after its last `open` or `ping`, compared across instances' clocks. A `ping` + for a row that is gone re-creates it, so a lost `open` heals on the next + heartbeat. ## Related diff --git a/services/connection_gateway/src/service/connection.rs b/services/connection_gateway/src/service/connection.rs index 6fbc8f14019..ff106d9d539 100644 --- a/services/connection_gateway/src/service/connection.rs +++ b/services/connection_gateway/src/service/connection.rs @@ -56,12 +56,13 @@ pub trait ConnectionRepo: Send + Sync { /// Deletes a [StoredConnectionEntity] from the database async fn remove_entity(&self, entity: &EntityConnection<'_>) -> anyhow::Result<()>; - /// Updates the last_ping timestamp for a given connection entity + /// Updates the last_ping timestamp for a given connection entity, or + /// returns `None` when there is no entry for it async fn update_last_entity_ping( &self, entity: &EntityConnection<'_>, timestamp: u64, - ) -> anyhow::Result; + ) -> anyhow::Result>; /// Updates the last_ping timestamp for a given top level connection async fn update_user_connection_last_ping( @@ -182,23 +183,40 @@ impl ConnectionManager { /// Updates the last_ping timestamp for a given connection entity /// This also propogates to the last_ping timestamp for the top level user connection + /// + /// A ping for an entity with no entry re-creates it: the client is still + /// watching, so a lost `open` (a failed write, a row cleared under a live + /// connection) must not hide the user until they reopen the entity. #[tracing::instrument(err, skip(self))] - pub async fn refresh_connection_entity(&self, entity: &EntityConnection<'_>) -> Result<()> { + pub async fn refresh_connection_entity( + &self, + connection_entity: UserEntityConnection<'_>, + ) -> Result<()> { let updated_timestamp = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap() .as_millis() as u64; - let connection_id = entity.connection_id.clone(); + let connection_id = connection_entity.extra.connection_id.to_string(); + let user_id = connection_entity.user_id.to_string(); - // Update the last_ping on the specific entity associated with the connection_id - let entity = self + let refreshed = self .persistence - .update_last_entity_ping(entity, updated_timestamp) + .update_last_entity_ping(&connection_entity.extra, updated_timestamp) .await?; + if refreshed.is_none() { + tracing::warn!( + connection_id = %connection_id, + entity_type = %connection_entity.extra.extra.entity_type, + entity_id = %connection_entity.extra.extra.entity_id, + "ping for an untracked entity; re-creating its entry" + ); + self.persistence + .insert_connection_entry(connection_entity) + .await?; + } - // // Update the user connection last_ping self.persistence - .update_user_connection_last_ping(&connection_id, &entity.user_id, updated_timestamp) + .update_user_connection_last_ping(&connection_id, &user_id, updated_timestamp) .await?; Ok(()) diff --git a/services/connection_gateway/src/service/connection/test.rs b/services/connection_gateway/src/service/connection/test.rs index f368a50eb09..8cc1d71d164 100644 --- a/services/connection_gateway/src/service/connection/test.rs +++ b/services/connection_gateway/src/service/connection/test.rs @@ -1,4 +1,5 @@ use super::*; +use crate::model::tracking::EntityConnectionExt; struct UnusedRepo; @@ -51,7 +52,7 @@ impl ConnectionRepo for UnusedRepo { &self, _entity: &EntityConnection<'_>, _timestamp: u64, - ) -> anyhow::Result { + ) -> anyhow::Result> { unimplemented!() } @@ -195,3 +196,113 @@ async fn binary_delivery_drops_a_full_connection_without_blocking_the_relay() { assert!(!manager.has_connection("connection")); assert!(forwarder.await.unwrap_err().is_cancelled()); } + +/// Holds no entity entries, so every entity ping misses, and records what +/// the manager writes in response. +#[derive(Default)] +struct EmptyEntityRepo { + inserted: std::sync::Mutex>, + user_pings: std::sync::Mutex>, +} + +#[async_trait] +impl ConnectionRepo for EmptyEntityRepo { + async fn insert_connection_entry( + &self, + connection: UserEntityConnection<'_>, + ) -> anyhow::Result { + let stored = StoredConnectionEntity::from(connection); + self.inserted.lock().unwrap().push(stored.clone()); + Ok(stored) + } + + async fn get_entries_by_entity( + &self, + _entity: &Entity<'_>, + ) -> anyhow::Result> { + unimplemented!() + } + + async fn get_entries_by_connection_id( + &self, + _connection_id: &str, + ) -> anyhow::Result> { + unimplemented!() + } + + async fn get_connection(&self, _connection_id: &str) -> anyhow::Result { + unimplemented!() + } + + async fn get_entry_for_connection_entity( + &self, + _entity: EntityConnection<'_>, + ) -> anyhow::Result> { + unimplemented!() + } + + async fn remove_all_entries_for_by_connection_id( + &self, + _connection_id: &str, + ) -> anyhow::Result<()> { + unimplemented!() + } + + async fn remove_entity(&self, _entity: &EntityConnection<'_>) -> anyhow::Result<()> { + unimplemented!() + } + + async fn update_last_entity_ping( + &self, + _entity: &EntityConnection<'_>, + _timestamp: u64, + ) -> anyhow::Result> { + Ok(None) + } + + async fn update_user_connection_last_ping( + &self, + connection_id: &str, + user: &str, + _timestamp: u64, + ) -> anyhow::Result<()> { + self.user_pings + .lock() + .unwrap() + .push((connection_id.to_owned(), user.to_owned())); + Ok(()) + } +} + +#[tokio::test] +async fn a_ping_for_an_entity_with_no_entry_recreates_it() { + let repo = Arc::new(EmptyEntityRepo::default()); + let manager = ConnectionManager { + connections: Arc::new(DashMap::new()), + connection_count: Arc::new(AtomicUsize::new(0)), + persistence: repo.clone(), + }; + + manager + .refresh_connection_entity( + model_entity::EntityType::Document + .with_entity_str("document-id") + .with_connection_str("connection-id") + .with_user_str("macro|viewer@example.com"), + ) + .await + .unwrap(); + + let inserted = repo.inserted.lock().unwrap(); + assert_eq!(inserted.len(), 1, "the missing entry must be written back"); + assert_eq!(inserted[0].pk, "#document#document-id"); + assert_eq!(inserted[0].sk, "connection-id"); + assert_eq!(inserted[0].user_id, "macro|viewer@example.com"); + assert_eq!( + *repo.user_pings.lock().unwrap(), + vec![( + "connection-id".to_owned(), + "macro|viewer@example.com".to_owned() + )] + ); +} diff --git a/services/connection_gateway/src/service/dynamodb.rs b/services/connection_gateway/src/service/dynamodb.rs index f6119f71c8b..a7151a8842c 100644 --- a/services/connection_gateway/src/service/dynamodb.rs +++ b/services/connection_gateway/src/service/dynamodb.rs @@ -120,6 +120,10 @@ impl ConnectionRepo for DynamoDbConnectionGatewayPersistence { .query() .table_name(self.table_exists.as_ref()) .key_condition_expression("PK = :primary_key") // Just the hash key condition + // Read right after the open, ping, or close that prompted it; an + // eventually consistent read can miss that write and send viewers + // a list without the user who just opened. + .consistent_read(true) .expression_attribute_values( ":primary_key", AttributeValue::S(format!("#{}#{}", entity.entity_type, entity.entity_id)), @@ -279,7 +283,7 @@ impl ConnectionRepo for DynamoDbConnectionGatewayPersistence { &self, entity: &EntityConnection<'_>, timestamp: u64, - ) -> Result { + ) -> Result> { let pk = format!("#{}#{}", entity.extra.entity_type, entity.extra.entity_id); let sk = entity.connection_id.to_string(); @@ -287,7 +291,7 @@ impl ConnectionRepo for DynamoDbConnectionGatewayPersistence { names.insert("#last_ping".to_string(), "last_ping".to_string()); names.insert("#entity_type".to_string(), "entity_type".to_string()); - let res = self + let res = match self .client .update_item() .table_name(self.table_exists.as_ref()) @@ -300,17 +304,16 @@ impl ConnectionRepo for DynamoDbConnectionGatewayPersistence { .return_values(ReturnValue::AllNew) .send() .await - .map_err(|e| { - if e.as_service_error() - .and_then(|se| se.code()) - .map(|code| code == "ConditionalCheckFailedException") - .unwrap_or(false) - { - anyhow::anyhow!("Connection entity no longer exists, skipping ping update") - } else { - anyhow::anyhow!("Failed to update last_ping: {}", e) - } - })?; + { + Ok(res) => res, + Err(e) + if e.as_service_error().and_then(|se| se.code()) + == Some("ConditionalCheckFailedException") => + { + return Ok(None); + } + Err(e) => anyhow::bail!("Failed to update last_ping: {}", e), + }; let attrs = res .attributes @@ -319,7 +322,7 @@ impl ConnectionRepo for DynamoDbConnectionGatewayPersistence { let entity: StoredConnectionEntity = serde_dynamo::from_item(attrs).context("Unable to deserialize entity")?; - Ok(entity) + Ok(Some(entity)) } async fn update_user_connection_last_ping( diff --git a/services/connection_gateway/src/service/tracker.rs b/services/connection_gateway/src/service/tracker.rs index 2e6fdcbfd04..bc15e5abaa1 100644 --- a/services/connection_gateway/src/service/tracker.rs +++ b/services/connection_gateway/src/service/tracker.rs @@ -93,7 +93,7 @@ pub async fn track_entity( TrackAction::Ping => { ctx.api_context .connection_manager - .refresh_connection_entity(&data.entity.extra) + .refresh_connection_entity(data.entity.clone()) .await .ok(); } From 6cc31ddfb0352975807955d62f1afc530e12d8a5 Mon Sep 17 00:00:00 2001 From: Wolf Mermelstein Date: Wed, 28 Oct 2026 12:20:26 +0000 Subject: [PATCH 5/5] test(agent_harness_service): follow the optional ping result in the repo stub Why: ConnectionRepo::update_last_entity_ping now returns None for a missing row, and the harness test stub still had the old signature. What: update the stub's return type. --- .../agent_harness_service/src/cognition/api/context/test.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/services/agent_harness_service/src/cognition/api/context/test.rs b/services/agent_harness_service/src/cognition/api/context/test.rs index be6a2e4a4df..a23d8b65a1d 100644 --- a/services/agent_harness_service/src/cognition/api/context/test.rs +++ b/services/agent_harness_service/src/cognition/api/context/test.rs @@ -119,7 +119,7 @@ impl ConnectionRepo for MockConnectionRepo { &self, _entity: &EntityConnection<'_>, _timestamp: u64, - ) -> anyhow::Result { + ) -> anyhow::Result> { unimplemented!() } async fn update_user_connection_last_ping(