fix me

Author: jherrera-jump

book/api/websocket.md

@@ -597,6 +597,9 @@ stake that is activating/deactivating in the current epoch and any stake
 that was explicilty undelegated prior to restart (e.g. inactive testnet
 participants or bad actors).
 
+During the `waiting_for_supermajority` phase, per-peer offline status
+is available via the [`wfs_offline_peers`](#wfs_offline_peers) topic.
+
 #### `summary.schedule_strategy`
 | frequency  | type     | example |
 |------------|----------|---------|
@@ -1858,6 +1861,60 @@ full and includes this node itself, nodes with a different
 `shred_version`, nodes publishing corrupt or bad information, and so
 on.
 
+### wfs_offline_peers
+The `wfs_offline_peers` topic publishes edge-triggered add/remove
+messages for staked peers that are **not** currently active on gossip.
+
+The message is only published if the client is configured to boot with
+the `waiting_for_supermajority` phase enabled.
+
+#### `wfs_offline_peers.update`
+| frequency        | type                              | example |
+|------------------|-----------------------------------|---------|
+| *Once* + *Live*  | `ClusterRestartOfflinePeerUpdate` | below   |
+
+::: details Example
+
+**Initial state (all peers offline):**
+```json
+{
+    "topic": "wfs_offline_peers",
+    "key": "update",
+    "value": {
+        "add": [
+            {
+                "identity": "Fe4StcZSQ228dKK2hni7aCP7ZprNhj8QKWzFe5usGFYF",
+                "stake": "5812",
+                "info": null
+            }
+        ],
+        "remove": [
+          { "identity": "9aE6Bp1hbDpMFKqnWGUMbfxfMPXswPbkNwNrSjhpFiSN" }
+        ]
+    }
+}
+```
+
+:::
+
+**`ClusterRestartOfflinePeerUpdate`**
+| Field           | Type                   | Description |
+|-----------------|------------------------|-------------|
+| identity        | `string`               | Identity public key of the validator, encoded in base58 |
+| stake           | `string`               | Total stake in lamports for this identity (aggregated across all vote accounts), derived from the snapshot manifest |
+| info            | `PeerUpdateInfo\|null` | Self-reported validator information from the ConfigProgram, if available, and `null` otherwise |
+
+**`ClusterRestartOfflinePeerRemove`**
+| Field    | Type     | Description |
+|----------|----------|-------------|
+| identity | `string` | Identity public key of the validator, encoded in base58 |
+
+**`ClusterRestartOfflinePeersUpdate`**
+| Field  | Type                                | Description |
+|--------|-------------------------------------|-------------|
+| add    | `ClusterRestartOfflinePeerUpdate[]` | List of peers that became offline (or all offline peers on initial connect) |
+| remove | `ClusterRestartOfflinePeerRemove[]` | List of peers that came back online and are no longer offline |
+
 ### slot
 Slots are opportunities for a leader to produce a block. A slot can be
 in one of five levels, and in typical operation a slot moves through

src/disco/gui/fd_gui_peers.c

@@ -4,6 +4,7 @@
 #include "fd_gui_metrics.h"
 
 #include "../../disco/metrics/fd_metrics_base.h"
+#include "../../disco/shred/fd_stake_ci.h"
 
 FD_IMPORT_BINARY( dbip_f, "src/disco/gui/dbip.bin.zst" );
 
@@ -270,6 +271,9 @@ fd_gui_peers_new( void *             shmem,
     build_geoip_trie( ctx, _dbip_nodes,   (uchar *)dbip_f,   dbip_f_sz,   &ctx->dbip,   FD_GUI_GEOIP_DBIP_MAX_NODES   );
 #endif
 
+    ctx->wfs_peers_cnt = 0UL;
+    wfs_fresh_dlist_join( wfs_fresh_dlist_new( ctx->wfs_fresh_dlist ) );
+
     return shmem;
 }
 
@@ -630,6 +634,55 @@ geoip_lookup( fd_gui_ip_db_t const * ip_db,
 
 #endif
 
+#define SORT_NAME wfs_peer_sort
+#define SORT_KEY_T fd_gui_wfs_offline_peer_t
+#define SORT_BEFORE(a,b) (memcmp( (a).identity_key.uc, (b).identity_key.uc, 32UL )<0)
+#include "../../util/tmpl/fd_sort.c"
+
+static void
+wfs_handle_contact_info_update( fd_gui_peers_ctx_t * peers,
+                                fd_pubkey_t const *  identity,
+                                long                 wallclock_nanos ) {
+  if( FD_LIKELY( !peers->wfs_peers_cnt ) ) return;
+
+  ulong idx = wfs_peer_sort_split( peers->wfs_peers, peers->wfs_peers_cnt, (fd_gui_wfs_offline_peer_t){ .identity_key = *identity } );
+  if( FD_UNLIKELY( idx>=peers->wfs_peers_cnt || memcmp( identity->uc, peers->wfs_peers[ idx ].identity_key.uc, sizeof(fd_pubkey_t) ) ) ) return;
+
+  fd_gui_wfs_offline_peer_t * wp = &peers->wfs_peers[ idx ];
+  wp->wallclock_nanos = wallclock_nanos;
+
+  if( !wp->is_online ) {
+    wp->is_online = 1;
+    wfs_fresh_dlist_idx_push_tail( peers->wfs_fresh_dlist, idx, peers->wfs_peers );
+
+    fd_gui_peers_printf_wfs_offline_update( peers, NULL, 0UL, &idx, 1UL );
+    fd_http_server_ws_broadcast( peers->http );
+  } else {
+    wfs_fresh_dlist_idx_remove( peers->wfs_fresh_dlist, idx, peers->wfs_peers );
+    wfs_fresh_dlist_idx_push_tail( peers->wfs_fresh_dlist, idx, peers->wfs_peers );
+  }
+}
+
+static void
+wfs_handle_contact_info_remove( fd_gui_peers_ctx_t * peers,
+                                fd_pubkey_t const *  identity ) {
+  if( FD_LIKELY( !peers->wfs_peers_cnt ) ) return;
+
+  ulong idx = wfs_peer_sort_split( peers->wfs_peers, peers->wfs_peers_cnt,
+      (fd_gui_wfs_offline_peer_t){ .identity_key = *identity } );
+  if( FD_UNLIKELY( idx>=peers->wfs_peers_cnt ||
+      memcmp( identity->uc, peers->wfs_peers[ idx ].identity_key.uc, 32UL ) ) ) return;
+
+  fd_gui_wfs_offline_peer_t * wp = &peers->wfs_peers[ idx ];
+  if( wp->is_online ) {
+    wfs_fresh_dlist_idx_remove( peers->wfs_fresh_dlist, idx, peers->wfs_peers );
+    wp->is_online = 0;
+
+    fd_gui_peers_printf_wfs_offline_update( peers, &idx, 1UL, NULL, 0UL );
+    fd_http_server_ws_broadcast( peers->http );
+  }
+}
+
 void
 fd_gui_peers_handle_gossip_update( fd_gui_peers_ctx_t *               peers,
                                    fd_gossip_update_message_t const * update,
@@ -655,7 +708,7 @@ fd_gui_peers_handle_gossip_update( fd_gui_peers_ctx_t *               peers,
             /* A new pubkey is not allowed to overwrite an existing valid index */
             FD_LOG_ERR(( "invariant violation: peer->pubkey.uc=%s != update->origin=%s ", ci_pk, og_pk ));
           }
-          FD_TEST( peer==fd_gui_peers_node_pubkey_map_ele_query_const( peers->node_pubkey_map, (fd_pubkey_t * )update->origin, NULL, peers->contact_info_table ) );
+          FD_TEST( peer==fd_gui_peers_node_pubkey_map_ele_query_const( peers->node_pubkey_map, (fd_pubkey_t const * )update->origin, NULL, peers->contact_info_table ) );
           fd_gui_peers_node_t * peer_sock = fd_gui_peers_node_sock_map_ele_query( peers->node_sock_map, &peer->contact_info.sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ], NULL, peers->contact_info_table );
           int found = 0;
           for( fd_gui_peers_node_t * p = peer_sock; !!p; p=(fd_gui_peers_node_t *)fd_gui_peers_node_sock_map_ele_next_const( p, NULL, peers->contact_info_table ) ) {
@@ -669,6 +722,7 @@ fd_gui_peers_handle_gossip_update( fd_gui_peers_ctx_t *               peers,
           /* update does nothing */
           if( FD_UNLIKELY( fd_gui_peers_contact_info_eq( &peer->contact_info, update->contact_info->value ) ) ) {
             peer->wallclock_nanos = FD_MILLI_TO_NANOSEC( update->wallclock );
+            wfs_handle_contact_info_update( peers, (fd_pubkey_t const *)update->origin, FD_MILLI_TO_NANOSEC( update->wallclock ) );
             break;
           }
 
@@ -696,6 +750,8 @@ fd_gui_peers_handle_gossip_update( fd_gui_peers_ctx_t *               peers,
           /* broadcast update to WebSocket clients */
           fd_gui_peers_printf_nodes( peers, (int[]){ FD_GUI_PEERS_NODE_UPDATE }, (ulong[]){ update->contact_info->idx }, 1UL );
           fd_http_server_ws_broadcast( peers->http );
+
+          wfs_handle_contact_info_update( peers, (fd_pubkey_t const *)update->origin, FD_MILLI_TO_NANOSEC( update->wallclock ) );
         } else {
 #if LOGGING
           char _pk[ FD_BASE58_ENCODED_32_SZ ];
@@ -748,6 +804,8 @@ fd_gui_peers_handle_gossip_update( fd_gui_peers_ctx_t *               peers,
           /* broadcast update to WebSocket clients */
           fd_gui_peers_printf_nodes( peers, (int[]){ FD_GUI_PEERS_NODE_ADD }, (ulong[]){ update->contact_info->idx }, 1UL );
           fd_http_server_ws_broadcast( peers->http );
+
+          wfs_handle_contact_info_update( peers, (fd_pubkey_t const *)update->origin, FD_MILLI_TO_NANOSEC( update->wallclock ) );
         }
         break;
       }
@@ -764,7 +822,7 @@ fd_gui_peers_handle_gossip_update( fd_gui_peers_ctx_t *               peers,
 #ifdef FD_GUI_USE_HANDHOLDING
         /* invariant checks */
         FD_TEST( peer->valid ); /* Should have already been in the table */
-        FD_TEST( peer==fd_gui_peers_node_pubkey_map_ele_query_const( peers->node_pubkey_map, (fd_pubkey_t * )update->origin, NULL, peers->contact_info_table ) );
+        FD_TEST( peer==fd_gui_peers_node_pubkey_map_ele_query_const( peers->node_pubkey_map, (fd_pubkey_t const * )update->origin, NULL, peers->contact_info_table ) );
         fd_gui_peers_node_t * peer_sock = fd_gui_peers_node_sock_map_ele_query( peers->node_sock_map, &peer->contact_info.sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ], NULL, peers->contact_info_table );
         int found = 0;
         for( fd_gui_peers_node_t const * p = peer_sock; !!p; p=(fd_gui_peers_node_t const *)fd_gui_peers_node_sock_map_ele_next_const( p, NULL, peers->contact_info_table ) ) {
@@ -775,6 +833,8 @@ fd_gui_peers_handle_gossip_update( fd_gui_peers_ctx_t *               peers,
         }
         FD_TEST( found );
 #endif
+        wfs_handle_contact_info_remove( peers, (fd_pubkey_t *)update->origin );
+
         fd_gui_peers_live_table_idx_remove          ( peers->live_table,      update->contact_info_remove->idx, peers->contact_info_table );
         fd_gui_peers_bandwidth_tracking_idx_remove  ( peers->bw_tracking,     update->contact_info_remove->idx, peers->contact_info_table );
         fd_gui_peers_node_sock_map_idx_remove_fast  ( peers->node_sock_map,   update->contact_info_remove->idx, peers->contact_info_table );
@@ -1030,6 +1090,38 @@ fd_gui_peers_handle_config_account( fd_gui_peers_ctx_t *  peers,
   fd_gui_peers_node_info_map_ele_insert( peers->node_info_map, node_info, peers->node_info_pool );
 }
 
+void
+fd_gui_peers_handle_manifest( fd_gui_peers_ctx_t *           peers,
+                              fd_snapshot_manifest_t const * manifest ) {
+  fd_vote_stake_weight_t scratch[ FD_RUNTIME_MAX_VOTE_ACCOUNTS ];
+  ulong scratch_cnt = 0UL;
+  for( ulong i=0UL; i<manifest->vote_accounts_len; i++ ) {
+    if( FD_UNLIKELY( manifest->vote_accounts[ i ].stake==0UL ) ) continue;
+    fd_memcpy( scratch[ scratch_cnt ].id_key.uc,   manifest->vote_accounts[ i ].node_account_pubkey, 32UL );
+    fd_memcpy( scratch[ scratch_cnt ].vote_key.uc, manifest->vote_accounts[ i ].vote_account_pubkey, 32UL );
+    scratch[ scratch_cnt ].stake = manifest->vote_accounts[ i ].stake;
+    scratch_cnt++;
+  }
+
+  /* Mirrors gossip WFS logic */
+  fd_stake_weight_t id_weights[ FD_RUNTIME_MAX_VOTE_ACCOUNTS ];
+  ulong id_cnt = compute_id_weights_from_vote_weights( id_weights, scratch, scratch_cnt );
+
+  /* Restore invariant: sorted by identity key */
+  fd_stake_weight_key_sort_inplace( id_weights, id_cnt );
+
+  for( ulong i=0UL; i<id_cnt; i++ ) {
+    peers->wfs_peers[ i ].identity_key    = id_weights[ i ].key;
+    peers->wfs_peers[ i ].stake           = id_weights[ i ].stake;
+    peers->wfs_peers[ i ].is_online       = 0;
+    peers->wfs_peers[ i ].wallclock_nanos = 0L;
+    peers->wfs_peers[ i ].fresh_prev      = ULONG_MAX;
+    peers->wfs_peers[ i ].fresh_next      = ULONG_MAX;
+  }
+  peers->wfs_peers_cnt = id_cnt;
+  wfs_fresh_dlist_join( wfs_fresh_dlist_new( peers->wfs_fresh_dlist ) );
+}
+
 
 static void
 fd_gui_peers_viewport_snap( fd_gui_peers_ctx_t * peers, ulong ws_conn_id ) {
@@ -1296,6 +1388,24 @@ int
 fd_gui_peers_poll( fd_gui_peers_ctx_t * peers, long now ) {
   int did_work = 0;
 
+  ulong evicted_cnt = 0UL;
+  while( FD_UNLIKELY( peers->wfs_peers_cnt && !wfs_fresh_dlist_is_empty( peers->wfs_fresh_dlist, peers->wfs_peers ) ) ) {
+    ulong head_idx = wfs_fresh_dlist_idx_peek_head( peers->wfs_fresh_dlist, peers->wfs_peers );
+    fd_gui_wfs_offline_peer_t * oldest = &peers->wfs_peers[ head_idx ];
+    if( oldest->wallclock_nanos > now - 15L*1000L*1000L*1000L ) break;
+
+    wfs_fresh_dlist_idx_pop_head( peers->wfs_fresh_dlist, peers->wfs_peers );
+    oldest->is_online = 0;
+
+    peers->scratch.wfs_peers[ evicted_cnt++ ] = head_idx;
+    if( FD_UNLIKELY( evicted_cnt>=256UL ) ) break;
+  }
+  if( FD_UNLIKELY( evicted_cnt ) ) {
+    fd_gui_peers_printf_wfs_offline_update( peers, peers->scratch.wfs_peers, evicted_cnt, NULL, 0UL );
+    fd_http_server_ws_broadcast( peers->http );
+    return 1; /* preserve STEM_BURST */
+  }
+
   /* update client viewports in a round-robin */
   if( FD_UNLIKELY( fd_gui_peers_ws_conn_rr_advance( peers, now ) ) ) {
     FD_TEST( peers->client_viewports[ peers->active_ws_conn_id ].connected );
@@ -1396,7 +1506,7 @@ fd_gui_peers_poll( fd_gui_peers_ctx_t * peers, long now ) {
     fd_http_server_ws_broadcast( peers->http );
 
     peers->next_gossip_stats_update_nanos = now + (FD_GUI_PEERS_GOSSIP_STATS_UPDATE_INTERVAL_MILLIS * 1000000L);
-    did_work = 1;
+    return 1; /* preserve STEM_BURST */
   }
 
   return did_work;
@@ -1415,6 +1525,16 @@ fd_gui_peers_ws_open( fd_gui_peers_ctx_t *  peers,
 
   fd_gui_peers_printf_node_all( peers );
   FD_TEST( !fd_http_server_ws_send( peers->http, ws_conn_id ) );
+
+  /* Send initial WFS offline peer snapshot if in WFS phase */
+  if( FD_UNLIKELY( peers->wfs_peers_cnt ) ) {
+    ulong added_cnt = 0UL;
+    for( ulong i=0UL; i<peers->wfs_peers_cnt; i++ ) {
+      if( FD_UNLIKELY( peers->wfs_peers[ i ].is_online ) ) peers->scratch.wfs_peers[ added_cnt++ ] = i;
+    }
+    fd_gui_peers_printf_wfs_offline_update( peers, peers->scratch.wfs_peers, peers->wfs_peers_cnt, NULL, 0UL );
+    FD_TEST( !fd_http_server_ws_send( peers->http, ws_conn_id ) );
+  }
 }
 
 void

src/disco/gui/fd_gui_peers.h

@@ -21,6 +21,7 @@
 #include "../../flamenco/runtime/fd_runtime_const.h"
 
 #include "../../waltz/http/fd_http_server.h"
+#include "../../discof/restore/utils/fd_ssmsg.h"
 #include "../topo/fd_topo.h"
 
 #if FD_HAS_ZSTD
@@ -85,6 +86,23 @@ struct fd_gui_geoip_node {
 
 typedef struct fd_gui_geoip_node fd_gui_geoip_node_t;
 
+struct fd_gui_wfs_offline_peer {
+  fd_pubkey_t identity_key;
+  ulong       stake;
+  int         is_online;
+  long        wallclock_nanos;
+
+  ulong       fresh_prev;
+  ulong       fresh_next;
+};
+typedef struct fd_gui_wfs_offline_peer fd_gui_wfs_offline_peer_t;
+
+#define DLIST_NAME  wfs_fresh_dlist
+#define DLIST_ELE_T fd_gui_wfs_offline_peer_t
+#define DLIST_PREV  fresh_prev
+#define DLIST_NEXT  fresh_next
+#include "../../util/tmpl/fd_dlist.c"
+
 #define FD_GUI_PEERS_NODE_NOP    (0)
 #define FD_GUI_PEERS_NODE_ADD    (1)
 #define FD_GUI_PEERS_NODE_UPDATE (2)
@@ -393,13 +411,20 @@ struct fd_gui_peers_ctx {
       int   actions[ FD_CONTACT_INFO_TABLE_SIZE ];
       ulong idxs   [ FD_CONTACT_INFO_TABLE_SIZE ];
     };
+    struct {
+      ulong wfs_peers[ FD_RUNTIME_MAX_VOTE_ACCOUNTS ];
+    };
   } scratch;
 
 #if FD_HAS_ZSTD
   ZSTD_DCtx * zstd_dctx;
 #endif
 
   fd_gui_ip_db_t dbip;
+
+  fd_gui_wfs_offline_peer_t wfs_peers[ FD_RUNTIME_MAX_VOTE_ACCOUNTS ];
+  ulong                     wfs_peers_cnt;
+  wfs_fresh_dlist_t         wfs_fresh_dlist[ 1 ];
 };
 
 typedef struct fd_gui_peers_ctx fd_gui_peers_ctx_t;
@@ -474,6 +499,10 @@ fd_gui_peers_handle_config_account( fd_gui_peers_ctx_t *  peers,
                                     uchar const *         data,
                                     ulong                 sz );
 
+void
+fd_gui_peers_handle_manifest( fd_gui_peers_ctx_t *           peers,
+                              fd_snapshot_manifest_t const * manifest );
+
 /* fd_gui_peers_ws_message handles incoming websocket request payloads
    requesting peer-related responses.  ws_conn_id is the connection id
    of the requester.  data is a pointer to the start of the