forest: fix forest publish edge cases

emwang-jump · emwang-jump · commit 3c9334190732 · 2025-07-01T16:12:45.000Z
diff --git a/src/discof/forest/fd_forest.c b/src/discof/forest/fd_forest.c
@@ -438,10 +438,26 @@ fd_forest_publish( fd_forest_t * forest, ulong new_root_slot ) {
   fd_forest_ele_t * old_root_ele = fd_forest_pool_ele( pool, forest->root );
   fd_forest_ele_t * new_root_ele = ancestry_frontier_query( forest, new_root_slot );
 
-# if FD_FOREST_USE_HANDHOLDING
-  FD_TEST( new_root_ele );                            /* caller error - not found */
-  FD_TEST( new_root_ele->slot > old_root_ele->slot ); /* caller error - inval */
-# endif
+#if FD_FOREST_USE_HANDHOLDING
+  if( FD_LIKELY( new_root_ele ) ) {
+    FD_TEST( new_root_ele->slot > old_root_ele->slot ); /* caller error - inval */
+  }
+#endif
+
+  /* Edge case where if we haven't been getting repairs, and we have a
+     gap between the root and orphans. we publish forward to a slot that
+     we don't have. This only case this should be happening is when we
+     load a second incremental and that incremental slot lives in the
+     gap. In that case this isn't a bug, but we should be treating this
+     new root like the snapshot slot / init root. Should be happening
+     very rarely given a well-functioning repair.  */
+
+  if( FD_UNLIKELY( !new_root_ele ) ) {
+    new_root_ele = acquire( forest, new_root_slot );
+    new_root_ele->complete_idx = 0;
+    new_root_ele->buffered_idx = 0;
+    fd_forest_frontier_ele_insert( frontier, new_root_ele, pool );
+  }
 
   /* First, remove the previous root, and add it to a FIFO prune queue.
      head points to the queue head (initialized with old_root_ele). */
@@ -475,7 +491,56 @@ fd_forest_publish( fd_forest_t * forest, ulong new_root_slot ) {
   }
 
   new_root_ele->parent = null; /* unlink new root from parent */
-  forest->root     = fd_forest_ancestry_idx_query( ancestry, &new_root_slot, null, pool );
+  forest->root         = fd_forest_ancestry_idx_query( ancestry, &new_root_slot, null, pool );
+  forest->root         = fd_ulong_if( forest->root == null, fd_forest_frontier_idx_query( frontier, &new_root_slot, null, pool ), forest->root );
+
+  /* If there is nothing on the frontier, we have hit an edge case
+     during catching up where all of our frontiers were < the new root.
+     In that case we need to continue repairing from the new root, so
+     add it to the frontier. */
+
+  if( FD_UNLIKELY( fd_forest_frontier_iter_done( fd_forest_frontier_iter_init( frontier, pool ), frontier, pool ) ) ) {
+    fd_forest_frontier_ele_insert( frontier, new_root_ele, pool );
+    new_root_ele->complete_idx = 0;
+    new_root_ele->buffered_idx = 0;
+    advance_frontier( forest, new_root_ele->slot, 0 );
+  }
+
+  /* Lastly, cleanup orphans if there orphan heads < new_root_slot.
+     First, add any relevant orphans to the prune queue. FIXME: NEED TO REMOVE BEFORE MODIFYING NEXT PTR */
+
+  fd_forest_orphaned_t * orphaned = fd_forest_orphaned( forest );
+  head = NULL;
+  for( fd_forest_orphaned_iter_t iter = fd_forest_orphaned_iter_init( orphaned, pool );
+       !fd_forest_orphaned_iter_done( iter, orphaned, pool );
+       iter = fd_forest_orphaned_iter_next( iter, orphaned, pool ) ) {
+    fd_forest_ele_t * ele = fd_forest_orphaned_iter_ele( iter, orphaned, pool );
+    if( FD_UNLIKELY( ele->slot < new_root_slot ) ) {
+      if( FD_UNLIKELY( head ) ) {
+        head = ele;
+        head->next = null;
+        tail = ele;
+      } else {
+        tail->next = iter.ele_idx;
+        tail = fd_forest_pool_ele( pool, tail->next );
+        tail->next = null;
+      }
+    }
+  }
+
+  /* Now BFS and clean up children of these orphan heads */
+  while( head ) {
+    fd_forest_ele_t * child = fd_forest_pool_ele( pool, head->child );
+    while( FD_LIKELY( child ) ) { /* add children to prune queue first */
+      tail->next = fd_forest_pool_idx( pool, child );
+      tail = fd_forest_pool_ele( pool, tail->next );
+      tail->next = null;
+      child = fd_forest_pool_ele( pool, child->sibling );
+    }
+    fd_forest_orphaned_ele_remove( orphaned, &head->slot, NULL, pool ); /* remove myself */
+    head = fd_forest_pool_ele( pool, head->next ); /* FIFO pop */
+    fd_forest_pool_ele_release( pool, head ); /* free head */
+  }
   return new_root_ele;
 }
 
diff --git a/src/discof/repair/fd_fec_chainer.c b/src/discof/repair/fd_fec_chainer.c
@@ -108,9 +108,9 @@ fd_fec_chainer_init( fd_fec_chainer_t * chainer, ulong slot, uchar merkle_root[s
   FD_TEST( fd_fec_pool_free( chainer->pool ) );
   fd_fec_ele_t * root = fd_fec_pool_ele_acquire( chainer->pool );
   FD_TEST( root );
-  root->key           = slot << 32 | ( UINT_MAX-1 ); // maintain invariant that no fec_set_idx=UINT_MAX lives in pool_ele
+  root->key           = slot << 32 | ( 0 ); // maintain invariant that no fec_set_idx=UINT_MAX lives in pool_ele
   root->slot          = slot;
-  root->fec_set_idx   = UINT_MAX-1;
+  root->fec_set_idx   = 0;
   root->data_cnt      = 0;
   root->data_complete = 1;
   root->slot_complete = 1;
@@ -125,9 +125,10 @@ fd_fec_chainer_init( fd_fec_chainer_t * chainer, ulong slot, uchar merkle_root[s
      UINT_MAX -> UINT_MAX-1 -> get_pool_ele(UINT_MAX-1)*/
 
   fd_fec_parent_t * p = fd_fec_parents_insert( chainer->parents, slot << 32 | UINT_MAX );
-  p->parent_key       = (slot << 32) | ( UINT_MAX - 1 );
+  p->parent_key       = (slot << 32) | ( 0 );
 
   fd_fec_frontier_ele_insert( chainer->frontier, root, chainer->pool );
+  chainer->root_fec = fd_fec_pool_idx( chainer->pool, root );
   return root;
 }
 
@@ -146,9 +147,17 @@ fd_fec_chainer_query( fd_fec_chainer_t * chainer, ulong slot, uint fec_set_idx )
   return fec;
 }
 
+static fd_fec_ele_t *
+fd_fec_chainer_remove( fd_fec_chainer_t * chainer, ulong key ) {
+  fd_fec_ele_t * fec = fd_fec_frontier_ele_remove( chainer->frontier, &key, NULL, chainer->pool );
+  fec = fd_ptr_if( !fec, fd_fec_ancestry_ele_remove( chainer->ancestry, &key, NULL, chainer->pool ), fec );
+  fec = fd_ptr_if( !fec, fd_fec_orphaned_ele_remove( chainer->orphaned, &key, NULL, chainer->pool ), fec );
+  return fec;
+}
+
 static int
 is_last_fec( ulong key ){
-  return ( (uint)fd_ulong_extract( key, 0, 31 ) & UINT_MAX ) == UINT_MAX; // lol fix
+  return (uint)fd_ulong_extract( key, 0, 31 ) == UINT_MAX;
 }
 
 static void
@@ -328,3 +337,62 @@ fd_fec_chainer_insert( fd_fec_chainer_t * chainer,
 
   return ele;
 }
+
+void
+fd_fec_chainer_publish( fd_fec_chainer_t * chainer, ulong new_root_slot ) {
+
+  /* 1. Prune children of the old root */
+  fd_fec_ele_t * old_root = fd_fec_pool_ele( chainer->pool, chainer->root_fec );
+  fd_fec_ele_t * new_root = fd_fec_chainer_query( chainer, new_root_slot, 0 );
+  FD_TEST( old_root );
+  if( FD_UNLIKELY( !new_root ) ) {
+    /* Happens during second incremental snapshot load */
+
+    new_root = fd_fec_pool_ele_acquire( chainer->pool );
+    new_root->key           = new_root_slot << 32 | ( 0 ); // maintain invariant that no fec_set_idx=UINT_MAX lives in pool_ele
+    new_root->slot          = new_root_slot;
+    new_root->fec_set_idx   = 0;
+    new_root->data_cnt      = 0;
+    new_root->data_complete = 1;
+    new_root->slot_complete = 1;
+    new_root->parent_off    = 0;
+    memset( new_root->chained_merkle_root, 0, FD_SHRED_MERKLE_ROOT_SZ );
+
+    fd_fec_parent_t * p = fd_fec_parents_insert( chainer->parents, new_root_slot << 32 | UINT_MAX );
+    p->parent_key       = (new_root_slot << 32) | ( 0 );
+
+    fd_fec_frontier_ele_insert( chainer->frontier, new_root, chainer->pool );
+  }
+
+  fd_fec_queue_push_tail( chainer->queue, old_root->key );
+
+  while( FD_LIKELY( !fd_fec_queue_empty( chainer->queue ) ) ) {
+    ulong key = fd_fec_queue_pop_head( chainer->queue );
+    fd_fec_ele_t * ele = fd_fec_chainer_query( chainer, key >> 32, (uint)key );
+    if( FD_UNLIKELY( !ele ) ) continue;
+
+    if( FD_UNLIKELY( ele->slot_complete ) ) {
+      fd_fec_children_t * fec_children = fd_fec_children_query( chainer->children, ele->slot, NULL );
+      if( FD_UNLIKELY( fec_children ) ) {
+        for( ulong off = fd_slot_child_offs_const_iter_init( fec_children->child_offs );
+            !fd_slot_child_offs_const_iter_done( off );
+            off = fd_slot_child_offs_const_iter_next( fec_children->child_offs, off ) ) {
+          ulong child_slot = ele->slot + off;
+
+          if( FD_UNLIKELY( child_slot == new_root_slot ) ) continue;
+
+          fd_fec_queue_push_tail( chainer->queue, child_slot << 32 | 0 );
+        }
+      }
+    } else {
+      ulong child_key = (ele->slot << 32) | (ele->key + ele->data_cnt);
+      fd_fec_queue_push_tail( chainer->queue, child_key );
+    }
+
+    /* Remove ele from the chainer. */
+
+    fd_fec_ele_t * remove = fd_fec_chainer_remove( chainer, ele->key );
+    FD_TEST( remove == ele );
+  }
+  chainer->root_fec = fd_fec_pool_idx( chainer->pool, new_root );
+}
diff --git a/src/discof/repair/fd_fec_chainer.h b/src/discof/repair/fd_fec_chainer.h
@@ -267,6 +267,7 @@ struct __attribute__((aligned(128UL))) fd_fec_chainer {
   fd_fec_children_t  * children; /* map of slot->child_offs for fast O(1) querying */
   ulong              * queue;    /* queue of FEC keys for BFS chaining */
   fd_fec_out_t       * out;      /* queue of FEC keys to deliver to application */
+  ulong              root_fec;   /* pool idx of the root FEC set */
 };
 
 FD_PROTOTYPES_BEGIN
@@ -365,6 +366,11 @@ fd_fec_chainer_insert( fd_fec_chainer_t * chainer,
                        uchar const        merkle_root[static FD_SHRED_MERKLE_ROOT_SZ],
                        uchar const        chained_merkle_root[static FD_SHRED_MERKLE_ROOT_SZ] );
 
+/* fd_fec_chainer_publish prunes the fec tree when the wmk is updated. */
+
+void
+fd_fec_chainer_publish( fd_fec_chainer_t * chainer, ulong new_root );
+
 FD_PROTOTYPES_END
 
 #endif /* HEADER_fd_src_discof_repair_fd_fec_chainer_h */
diff --git a/src/discof/repair/fd_repair_tile.c b/src/discof/repair/fd_repair_tile.c
@@ -767,15 +767,17 @@ after_frag( fd_repair_tile_ctx_t * ctx,
 
     ulong wmark = fd_fseq_query( ctx->wmark );
     if( FD_UNLIKELY( fd_forest_root_slot( ctx->forest ) == ULONG_MAX ) ) {
+      FD_LOG_NOTICE(( "Forest initializing with root %lu", wmark ));
       fd_forest_init( ctx->forest, wmark );
       uchar mr[ FD_SHRED_MERKLE_ROOT_SZ ] = { 0 }; /* FIXME */
       fd_fec_chainer_init( ctx->fec_chainer, wmark, mr );
       FD_TEST( fd_forest_root_slot( ctx->forest ) != ULONG_MAX );
-      FD_LOG_NOTICE(( "Forest initialized with root %lu", wmark ));
       ctx->prev_wmark = wmark;
     }
+
     if( FD_UNLIKELY( ctx->prev_wmark < wmark ) ) {
       fd_forest_publish( ctx->forest, wmark );
+      fd_fec_chainer_publish( ctx->fec_chainer, wmark );
       ctx->prev_wmark  = wmark;
       // invalidate our repair iterator
       ctx->repair_iter = fd_forest_iter_init( ctx->forest );
@@ -841,6 +843,7 @@ after_frag( fd_repair_tile_ctx_t * ctx,
           ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
           reasm->cnt = out.fec_set_idx + out.data_cnt;
           fd_stem_publish( ctx->stem, REPLAY_OUT_IDX, sig, 0, 0, 0, tsorig, tspub );
+          FD_LOG_INFO(( "REPLAY out %lu %u %u %d", out.slot, out.fec_set_idx, cnt, out.slot_complete ));
           if( FD_UNLIKELY( out.slot_complete ) ) {
             fd_reasm_remove( ctx->reasm, reasm );
           }
diff --git a/src/discof/repair/test_fec_chainer.c b/src/discof/repair/test_fec_chainer.c
@@ -183,6 +183,77 @@ test_single_fec( fd_wksp_t * wksp ){
 
 }
 
+void
+test_publish( fd_wksp_t * wksp ){
+  ulong fec_max = 32;
+
+  void * mem = fd_wksp_alloc_laddr( wksp, fd_fec_chainer_align(), fd_fec_chainer_footprint( fec_max ), 1UL );
+  FD_TEST( mem );
+  fd_fec_chainer_t * chainer = fd_fec_chainer_join( fd_fec_chainer_new( mem, fec_max, 0UL ) );
+
+  uchar mr_root[FD_SHRED_MERKLE_ROOT_SZ] = { 1 };
+  fd_fec_ele_t * f0_64 = fd_fec_chainer_init( chainer, 1, mr_root );
+
+  FD_TEST( fd_fec_frontier_ele_query( chainer->frontier, &f0_64->key, NULL, chainer->pool ) == f0_64 );
+
+  /* Typical startup behavior, turbine orphan FECs added */
+  fd_fec_chainer_insert( chainer, 10, 0, 32, 1, 1, 1, mr_root, mr_root );
+  fd_fec_chainer_insert( chainer, 9, 0, 32, 1, 1, 1, mr_root, mr_root );
+
+  /* simulating no FECs chained, but a new root is published */
+  ulong new_root = 2UL << 32 | 0;
+  fd_fec_chainer_publish( chainer, 2 );
+
+  FD_TEST( fd_fec_frontier_ele_query( chainer->frontier, &new_root, NULL, chainer->pool ) );
+
+  /* Chain off of root slot for a bit */
+  fd_fec_chainer_insert( chainer, 3, 0, 32, 1, 1, 1, mr_root, mr_root );
+  fd_fec_chainer_insert( chainer, 4, 0, 32, 1, 1, 1, mr_root, mr_root );
+  ulong new_frontier = 4UL << 32 | 0;
+  FD_TEST( fd_fec_frontier_ele_query( chainer->frontier, &new_frontier, NULL, chainer->pool ) );
+
+  /* Publish to ancestor */
+  fd_fec_chainer_publish( chainer, 3 );
+  FD_TEST( fd_fec_frontier_ele_query( chainer->frontier, &new_frontier, NULL, chainer->pool ) );
+
+  /* Make a tree
+
+  3 - 4 - 8 - 9 - 10
+    \ 5 - 6 - 7
+
+  */
+
+  fd_fec_chainer_insert( chainer, 5, 0, 32, 1, 1, 2, mr_root, mr_root );
+  fd_fec_chainer_insert( chainer, 6, 0, 32, 1, 1, 1, mr_root, mr_root );
+  fd_fec_chainer_insert( chainer, 7, 0, 32, 1, 1, 1, mr_root, mr_root );
+  fd_fec_chainer_insert( chainer, 8, 0, 32, 1, 1, 4, mr_root, mr_root );
+
+  uint frontier_cnt = 0;
+  ulong frontier_keys[2] = { 7UL << 32 | 0, 10UL << 32 | 0 };
+  for( fd_fec_frontier_iter_t iter = fd_fec_frontier_iter_init( chainer->frontier, chainer->pool ); !fd_fec_frontier_iter_done( iter, chainer->frontier, chainer->pool ); iter = fd_fec_frontier_iter_next( iter, chainer->frontier, chainer->pool ) ){
+    frontier_cnt++;
+  }
+  FD_TEST( frontier_cnt == sizeof(frontier_keys) / sizeof(ulong) );
+  for( uint i = 0; i < sizeof(frontier_keys) / sizeof(ulong); i++ ){
+    FD_TEST( fd_fec_frontier_ele_query( chainer->frontier, &frontier_keys[i], NULL, chainer->pool ) );
+  }
+
+  /* Publish down the tree */
+  fd_fec_chainer_publish( chainer, 4 );
+  new_frontier = 10UL << 32 | 0;
+  FD_TEST( fd_fec_frontier_ele_query( chainer->frontier, &new_frontier, NULL, chainer->pool ) );
+
+  FD_TEST( fd_fec_chainer_query( chainer, 4, 0 ) );
+  FD_TEST( !fd_fec_chainer_query( chainer, 5, 0 ) );
+  FD_TEST( !fd_fec_chainer_query( chainer, 6, 0 ) );
+  FD_TEST( !fd_fec_chainer_query( chainer, 7, 0 ) );
+  FD_TEST( fd_fec_chainer_query( chainer, 8, 0 ) );
+  FD_TEST( fd_fec_chainer_query( chainer, 9, 0 ) );
+  FD_TEST( fd_fec_chainer_query( chainer, 10, 0 ) );
+
+  fd_wksp_free_laddr( fd_fec_chainer_delete( fd_fec_chainer_leave( chainer ) ) );
+}
+
 int
 main( int argc, char ** argv ) {
   fd_boot( &argc, &argv );
@@ -195,6 +266,7 @@ main( int argc, char ** argv ) {
 
   // test_fec_ordering( wksp );
   // test_single_fec( wksp );
+  test_publish( wksp );
 
   ulong sig = fd_disco_repair_replay_sig( 3508496, 1, 32, 128 );
   FD_TEST( fd_disco_repair_replay_sig_slot( sig ) == 3508496 );
diff --git a/src/discof/replay/fd_replay_tile.c b/src/discof/replay/fd_replay_tile.c
@@ -1656,6 +1656,7 @@ read_snapshot( void *              _ctx,
     /* TODO: If prefetching the manifest is enabled it leads to
        incorrect snapshot loads. This needs to be looked into. */
     if( strlen( incremental )>0UL ) {
+      FD_LOG_NOTICE(("LOADING INCREMENTAL SNAPSHOT"));
       uchar * tmp_mem = fd_spad_alloc_check( ctx->runtime_spad, fd_snapshot_load_ctx_align(), fd_snapshot_load_ctx_footprint() );
 
       fd_snapshot_load_ctx_t * tmp_snap_ctx = fd_snapshot_load_new( tmp_mem,
@@ -1696,13 +1697,15 @@ read_snapshot( void *              _ctx,
     /* If we don't have an incremental snapshot, load the manifest and the status cache and initialize
          the objects because we don't have these from the incremental snapshot. */
     if( strlen( incremental )<=0UL ) {
+      FD_LOG_NOTICE(("LOADING MANIFEST AND STATUS CACHE BECAUSE NO INCREMENTAL SNAPSHOT"));
       fd_snapshot_load_manifest_and_status_cache( snap_ctx, NULL,
         FD_SNAPSHOT_RESTORE_MANIFEST | FD_SNAPSHOT_RESTORE_STATUS_CACHE );
 
       kickoff_repair_orphans( ctx, stem );
       /* If we don't have an incremental snapshot, we can still kick off
          sending the stake weights and snapshot slot to repair. */
     } else {
+      FD_LOG_NOTICE(("LOADING MANIFEST AND STATUS CACHE W/ INCREMENTAL SNAPSHOT"));
       /* If we have an incremental snapshot, load the manifest and the status cache,
           and don't initialize the objects because we did this above from the incremental snapshot. */
       fd_snapshot_load_manifest_and_status_cache( snap_ctx, NULL, FD_SNAPSHOT_RESTORE_NONE );
@@ -1717,6 +1720,7 @@ read_snapshot( void *              _ctx,
 
     /* The slot of the full snapshot should be used as the base slot to verify the incremental snapshot,
        not the slot context's slot - which is the slot of the incremental, not the full snapshot. */
+    FD_LOG_NOTICE(( "LOADING ANOTHER INCREMENTAL SNAPSHOT"));
     fd_snapshot_load_all( incremental,
                           ctx->incremental_src_type,
                           NULL,
@@ -1885,10 +1889,8 @@ init_after_snapshot( fd_replay_tile_ctx_t * ctx,
        are no funk txns to publish, and all rooted slots have already
        been registered in the txncache when we loaded the snapshot. */
 
-    fd_funk_txn_xid_t xid = { .ul = { root, root } };
     if( FD_LIKELY( ctx->blockstore ) ) fd_blockstore_publish( ctx->blockstore, ctx->blockstore_fd, root );
     if( FD_LIKELY( ctx->forks ) ) fd_forks_publish( ctx->forks, root );
-    if( FD_LIKELY( ctx->funk ) ) funk_and_txncache_publish( ctx, root, &xid );
     if( FD_LIKELY( ctx->epoch_forks ) ) fd_epoch_forks_publish( ctx->epoch_forks, root );
 
     fd_fseq_update( ctx->published_wmark, root );
