refactor(reasm): overwrite bad CMRs using slot and parent_off

lidatong · lidatong · commit df592ac59ad0 · 2025-08-12T22:29:54.000Z
diff --git a/src/discof/forest/fd_forest.c b/src/discof/forest/fd_forest.c
@@ -892,8 +892,7 @@ ancestry_print3( fd_forest_t const * forest, fd_forest_ele_t const * ele, int sp
 
 void
 fd_forest_ancestry_print( fd_forest_t const * forest ) {
-  FD_LOG_NOTICE(("\n\n[Ancestry]\n\n" ) );
-
+  printf(("\n\n[Ancestry]\n" ) );
   ancestry_print3( forest, fd_forest_pool_ele_const( fd_forest_pool_const( forest ), forest->root ), 0, "[", NULL, 0 );
 }
 
@@ -913,7 +912,7 @@ fd_forest_frontier_print( fd_forest_t const * forest ) {
 
 void
 fd_forest_orphaned_print( fd_forest_t const * forest ) {
-  printf( "\n\n[Orphaned]\n" );
+  printf( "\n[Orphaned]\n" );
   fd_forest_orphaned_t const * orphaned = fd_forest_orphaned_const( forest );
   fd_forest_ele_t const * pool = fd_forest_pool_const( forest );
   for( fd_forest_orphaned_iter_t iter = fd_forest_orphaned_iter_init( orphaned, pool );
@@ -928,10 +927,11 @@ void
 fd_forest_print( fd_forest_t const * forest ) {
   if( FD_UNLIKELY( forest->root == ULONG_MAX ) ) return;
 # if FD_FOREST_PRINT
+  FD_LOG_NOTICE(("\n\n[Forest]" ) );
   fd_forest_ancestry_print( forest );
   fd_forest_frontier_print( forest );
   fd_forest_orphaned_print( forest );
-  printf("\n\n");
+  printf("\n");
 # endif
 }
 
diff --git a/src/discof/forest/test_forest.c b/src/discof/forest/test_forest.c
@@ -639,7 +639,7 @@ main( int argc, char ** argv ) {
   test_publish_incremental( wksp );
   test_out_of_order( wksp );
   test_forks( wksp );
-  // test_print_tree( wksp );
+  test_print_tree( wksp );
   // test_large_print_tree( wksp);
   test_linear_forest_iterator( wksp );
   test_branched_forest_iterator( wksp );
diff --git a/src/discof/reasm/fd_reasm.c b/src/discof/reasm/fd_reasm.c
@@ -8,6 +8,7 @@ fd_reasm_align( void ) {
 
 ulong
 fd_reasm_footprint( ulong fec_max ) {
+  int lgf_max = fd_ulong_find_msb( fd_ulong_pow2_up( fec_max ) );
   return FD_LAYOUT_FINI(
     FD_LAYOUT_APPEND(
     FD_LAYOUT_APPEND(
@@ -17,6 +18,7 @@ fd_reasm_footprint( ulong fec_max ) {
     FD_LAYOUT_APPEND(
     FD_LAYOUT_APPEND(
     FD_LAYOUT_APPEND(
+    FD_LAYOUT_APPEND(
     FD_LAYOUT_INIT,
       alignof(fd_reasm_t), sizeof(fd_reasm_t)            ),
       pool_align(),        pool_footprint    ( fec_max ) ),
@@ -26,6 +28,7 @@ fd_reasm_footprint( ulong fec_max ) {
       subtrees_align(),    subtrees_footprint( fec_max ) ),
       bfs_align(),         bfs_footprint     ( fec_max ) ),
       out_align(),         out_footprint     ( fec_max ) ),
+      slot_mr_align(),     slot_mr_footprint ( lgf_max ) ),
     fd_reasm_align() );
 }
 
@@ -56,6 +59,8 @@ fd_reasm_new( void * shmem, ulong fec_max, ulong seed ) {
 
   fd_memset( shmem, 0, footprint );
 
+  int lgf_max = fd_ulong_find_msb( fd_ulong_pow2_up( fec_max ) );
+
   fd_reasm_t * reasm;
   FD_SCRATCH_ALLOC_INIT( l, shmem );
   reasm           = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_reasm_t), sizeof(fd_reasm_t)            );
@@ -66,6 +71,7 @@ fd_reasm_new( void * shmem, ulong fec_max, ulong seed ) {
   void * subtrees = FD_SCRATCH_ALLOC_APPEND( l, subtrees_align(),    subtrees_footprint( fec_max ) );
   void * bfs      = FD_SCRATCH_ALLOC_APPEND( l, bfs_align(),         bfs_footprint     ( fec_max ) );
   void * out      = FD_SCRATCH_ALLOC_APPEND( l, out_align(),         out_footprint     ( fec_max ) );
+  void * slot_mr  = FD_SCRATCH_ALLOC_APPEND( l, slot_mr_align(),     slot_mr_footprint ( lgf_max ) );
   FD_TEST( FD_SCRATCH_ALLOC_FINI( l, fd_reasm_align() ) == (ulong)shmem + footprint );
 
   reasm->root     = pool_idx_null( pool );
@@ -76,6 +82,7 @@ fd_reasm_new( void * shmem, ulong fec_max, ulong seed ) {
   reasm->subtrees = subtrees_new ( subtrees, fec_max, seed );
   reasm->bfs      = bfs_new      ( bfs,      fec_max       );
   reasm->out      = out_new      ( out,      fec_max       );
+  reasm->slot_mr  = slot_mr_new  ( slot_mr,  lgf_max       );
 
   return shmem;
 }
@@ -96,6 +103,7 @@ fd_reasm_join( void * shreasm ) {
   reasm->subtrees = subtrees_join( reasm->subtrees );
   reasm->bfs      = bfs_join     ( reasm->bfs      );
   reasm->out      = out_join     ( reasm->out      );
+  reasm->slot_mr  = slot_mr_join ( reasm->slot_mr  );
 
   return reasm;
 }
@@ -150,7 +158,10 @@ fd_reasm_init( fd_reasm_t * reasm, fd_hash_t const * merkle_root, ulong slot ) {
   fec->fec_set_idx     = 0;
   fec->data_cnt        = 0;
   fec->data_complete   = 0;
-  fec->slot_complete   = 0;
+  fec->slot_complete   = 1;
+
+  slot_mr_t * slot_mr = slot_mr_insert( reasm->slot_mr, slot );
+  slot_mr->block_id   = fec->key;
 
   /* Set this dummy FEC as the root and add it to the frontier. */
 
@@ -188,6 +199,20 @@ fd_reasm_query( fd_reasm_t * reasm,
   return fec;
 }
 
+static void
+check_cmr( fd_reasm_t * reasm, fd_reasm_fec_t * child ) {
+  if( FD_UNLIKELY( child->fec_set_idx==0 && !fd_reasm_query( reasm, &child->cmr ) ) ) {
+    slot_mr_t * slot_mr_parent = slot_mr_query( reasm->slot_mr, child->slot - child->parent_off, NULL );
+    if( FD_LIKELY( slot_mr_parent ) ) {
+      fd_reasm_fec_t * parent = fd_reasm_query( reasm, &slot_mr_parent->block_id );
+      if( FD_LIKELY( parent ) ) {
+        FD_LOG_NOTICE(( "overwriting bad block_id for FEC slot: %lu fec_set_idx: %u from %s to %s", child->slot, child->fec_set_idx, FD_BASE58_ENC_32_ALLOCA( &child->cmr ), FD_BASE58_ENC_32_ALLOCA( &parent->key ) ));
+        child->cmr = parent->key; /* use the parent's merkle root */
+      }
+    }
+  }
+}
+
 static void
 link( fd_reasm_t     * reasm,
       fd_reasm_fec_t * parent,
@@ -212,7 +237,7 @@ fd_reasm_insert( fd_reasm_t *      reasm,
                  ushort            data_cnt,
                  int               data_complete,
                  int               slot_complete ) {
-  // FD_LOG_NOTICE(( "inserting %s %lu %u %u %d %d", FD_BASE58_ENC_32_ALLOCA( merkle_root ), slot, fec_set_idx, data_cnt, data_complete, slot_complete ));
+  // FD_LOG_NOTICE(( "inserting (%lu %u) %s. %u %d %d", slot, fec_set_idx, FD_BASE58_ENC_32_ALLOCA( merkle_root ), data_cnt, data_complete, slot_complete ));
 
 # if FD_REASM_USE_HANDHOLDING
   FD_TEST( pool_free( reasm->pool ) );
@@ -244,6 +269,29 @@ fd_reasm_insert( fd_reasm_t *      reasm,
   fec->data_complete   = data_complete;
   fec->slot_complete   = slot_complete;
 
+  /* This is a gross case reasm needs to handle because Agave currently
+     does not validate chained merkle roots across slots ie. if a leader
+     sends a bad chained merkle root on a slot boundary, the cluster might
+     converge on the leader's block anyways.  So we overwrite the chained
+     merkle root based on the slot and parent_off metadata.  There are two
+     cases: 1. we receive the parent before the child.  In this case we
+     just overwrite the child's CMR.  2. we receive the child before the
+     parent.  In this case every time we receive a new FEC set we need to
+     check the orphan roots for whether we can link the orphan to the new
+     FEC via slot metadata, since the chained merkle root metadata on that
+     orphan root might be wrong. */
+
+  if( FD_UNLIKELY( slot_complete ) ) {
+    slot_mr_t * slot_mr = slot_mr_query( reasm->slot_mr, slot, NULL );
+    if( FD_UNLIKELY( slot_mr ) ) {
+      FD_LOG_WARNING(( "equivocating block_id for FEC slot: %lu fec_set_idx: %u prev: %s curr: %s", fec->slot, fec->fec_set_idx, FD_BASE58_ENC_32_ALLOCA( &slot_mr->block_id ), FD_BASE58_ENC_32_ALLOCA( &fec->key ) )); /* it's possible there's equivocation... */
+    } else {
+      slot_mr = slot_mr_insert( reasm->slot_mr, slot );
+      slot_mr->block_id   = fec->key;
+    }
+  }
+  check_cmr( reasm, fec ); /* handle receiving parent before child */
+
   /* First, we search for the parent of this new FEC and link if found.
      The new FEC set may result in a new leaf or a new orphan tree root
      so we need to check that. */
@@ -286,6 +334,7 @@ fd_reasm_insert( fd_reasm_t *      reasm,
   }
   while( FD_LIKELY( !bfs_empty( bfs ) ) ) {
     fd_reasm_fec_t * orphan_root = pool_ele( reasm->pool, bfs_pop_head( bfs ) );
+    check_cmr( reasm, orphan_root ); /* handle receiving child before parent */
     if( FD_LIKELY( orphan_root && 0==memcmp( orphan_root->cmr.uc, fec->key.uc, sizeof(fd_hash_t) ) ) ) { /* this orphan_root is a direct child of fec */
       link( reasm, fec, orphan_root );
       if( FD_UNLIKELY( is_root ) ) { /* this is an orphan tree */
@@ -367,6 +416,10 @@ fd_reasm_publish( fd_reasm_t * reasm, fd_hash_t const * merkle_root ) {
     }
     fd_reasm_fec_t * next = pool_ele( pool, head->next ); /* pophead */
     pool_ele_release( pool, head );                       /* release */
+
+    slot_mr_t * slot_mr = slot_mr_query( reasm->slot_mr, head->slot, NULL );
+    if( FD_UNLIKELY( slot_mr ) ) slot_mr_remove( reasm->slot_mr, slot_mr );
+
     head = next;                                          /* advance */
   }
   newr->parent = null;                   /* unlink old root */
@@ -376,7 +429,7 @@ fd_reasm_publish( fd_reasm_t * reasm, fd_hash_t const * merkle_root ) {
 
 #include <stdio.h>
 
-static void
+FD_FN_UNUSED static void
 print( fd_reasm_t const * reasm, fd_reasm_fec_t const * fec, int space, const char * prefix ) {
   fd_reasm_fec_t * pool = reasm->pool;
 
@@ -402,7 +455,32 @@ print( fd_reasm_t const * reasm, fd_reasm_fec_t const * fec, int space, const ch
 
 void
 fd_reasm_print( fd_reasm_t const * reasm ) {
-  FD_LOG_NOTICE( ( "\n\n[reasm]" ) );
-  print( reasm, pool_ele_const( reasm->pool, reasm->root ), 0, "" );
+  FD_LOG_NOTICE( ( "\n\n[Reasm]" ) );
+  fd_reasm_fec_t * pool = reasm->pool;
+
+  printf(("\n\n[Frontier]\n" ) );
+  frontier_t * frontier = reasm->frontier;
+  for( frontier_iter_t iter = frontier_iter_init(       frontier, pool );
+                             !frontier_iter_done( iter, frontier, pool );
+                       iter = frontier_iter_next( iter, frontier, pool ) ) {
+    fd_reasm_fec_t const * fec = pool_ele_const( reasm->pool, iter.ele_idx );
+    printf( "(%lu, %u) %s\n", fec->slot, fec->fec_set_idx, FD_BASE58_ENC_32_ALLOCA( &fec->key ) );
+  }
+
+  printf(("\n\n[Subtrees]\n" ) );
+  subtrees_t * subtrees = reasm->subtrees;
+  for( subtrees_iter_t iter = subtrees_iter_init(       subtrees, pool );
+                             !subtrees_iter_done( iter, subtrees, pool );
+                       iter = subtrees_iter_next( iter, subtrees, pool ) ) {
+    fd_reasm_fec_t const * fec = pool_ele_const( reasm->pool, iter.ele_idx );
+    printf( "(%lu, %u) %s\n", fec->slot, fec->fec_set_idx, FD_BASE58_ENC_32_ALLOCA( &fec->key ) );
+  }
+
+  // print( reasm, pool_ele_const( reasm->pool, reasm->root ), 0, "" );
+  // printf( "\n\n" );
+  // for( out_iter_t iter = out_iter_init( reasm->out ); !out_iter_done( reasm->out, iter ); iter = out_iter_next( reasm->out, iter ) ) {
+  //   ulong * idx = out_iter_ele( reasm->out, iter );
+  //   printf( "%s\n", FD_BASE58_ENC_32_ALLOCA( pool_ele_const( reasm->pool, *idx ) ) );
+  // }
   printf( "\n\n" );
 }
diff --git a/src/discof/reasm/fd_reasm_private.h b/src/discof/reasm/fd_reasm_private.h
@@ -40,6 +40,19 @@
 #define DEQUE_T                ulong
 #include "../../util/tmpl/fd_deque_dynamic.c"
 
+typedef struct {
+  ulong     slot;
+  fd_hash_t block_id;
+} slot_mr_t;
+
+#define MAP_NAME         slot_mr
+#define MAP_T            slot_mr_t
+#define MAP_KEY          slot
+#define MAP_KEY_NULL     ULONG_MAX
+#define MAP_KEY_INVAL(k) ((k)==MAP_KEY_NULL)
+#define MAP_MEMOIZE      0
+#include "../../util/tmpl/fd_map_dynamic.c"
+
 struct __attribute__((aligned(128UL))) fd_reasm {
   ulong            root;     /* pool idx of the root FEC set */
   ulong            slot0;    /* special initialization slot. chains first FEC */
@@ -50,4 +63,5 @@ struct __attribute__((aligned(128UL))) fd_reasm {
   subtrees_t *     subtrees; /* map of mr->fec. roots of the orphaned subtrees */
   ulong *          bfs;      /* internal queue of pool idxs for BFS */
   ulong *          out;      /* delivery queue of pool idxs to output */
+  slot_mr_t *      slot_mr;  /* map of slot->mr */
 };
diff --git a/src/discof/reasm/test_reasm.c b/src/discof/reasm/test_reasm.c
@@ -107,7 +107,7 @@ test_insert( fd_wksp_t * wksp ) {
 }
 
 void
-test_publish( fd_wksp_t * wksp ){
+test_publish( fd_wksp_t * wksp ) {
   ulong        fec_max = 32;
   void *       mem     = fd_wksp_alloc_laddr( wksp, fd_reasm_align(), fd_reasm_footprint( fec_max ), 1UL );
   fd_reasm_t * reasm   = fd_reasm_join( fd_reasm_new( mem, fec_max, 0UL ) );
@@ -181,6 +181,44 @@ test_publish( fd_wksp_t * wksp ){
   fd_wksp_free_laddr( fd_reasm_delete( fd_reasm_leave( reasm ) ) );
 }
 
+void
+test_slot_mr( fd_wksp_t * wksp ) {
+  ulong        fec_max = 32;
+  void *       mem     = fd_wksp_alloc_laddr( wksp, fd_reasm_align(), fd_reasm_footprint( fec_max ), 1UL );
+  fd_reasm_t * reasm   = fd_reasm_join( fd_reasm_new( mem, fec_max, 0UL ) );
+  FD_TEST( reasm );
+  FD_TEST( reasm->slot_mr );
+
+  fd_reasm_fec_t * pool = reasm->pool;
+  // ulong            null = pool_idx_null( pool );
+
+  ancestry_t * ancestry = reasm->ancestry;
+  frontier_t * frontier = reasm->frontier;
+  subtrees_t * subtrees = reasm->subtrees;
+
+  fd_hash_t mr0[1] = {{{ 0 }}};
+  fd_hash_t mr1[1] = {{{ 1 }}};
+  fd_hash_t mr2[1] = {{{ 2 }}};
+  fd_hash_t mr3[1] = {{{ 3 }}};
+  fd_hash_t mr4[1] = {{{ 4 }}};
+
+  fd_reasm_init( reasm, mr1, 1 );
+  fd_reasm_fec_t * fec1 = fd_reasm_query( reasm, mr1 );
+  FD_TEST( fec1 );
+  FD_TEST( frontier_ele_query( frontier, &fec1->key, NULL, pool ) == fec1 );
+
+  fd_reasm_fec_t * fec2 = fd_reasm_insert( reasm, mr2, mr0, 2, 0, 1, 32, 0, 1 ); /* insert with bad mr0 should be mr1 */
+  FD_TEST( ancestry_ele_query( ancestry, &fec1->key, NULL, pool ) );             /* successfully chains anyways */
+  FD_TEST( frontier_ele_query( frontier, &fec2->key, NULL, pool ) );
+
+  fd_reasm_fec_t * fec4 = fd_reasm_insert( reasm, mr4, mr0, 4, 0, 1, 32, 0, 1 ); /* insert with bad mr0 should be mr3 */
+  FD_TEST( subtrees_ele_query( subtrees, &fec4->key, NULL, pool ) );             /* orphaned */
+
+  fd_reasm_fec_t * fec3 = fd_reasm_insert( reasm, mr3, mr2, 3, 0, 1, 32, 0, 1 );
+  FD_TEST( ancestry_ele_query( ancestry, &fec3->key, NULL, pool ) );             /* mr3 should chain to 2 */
+  FD_TEST( frontier_ele_query( frontier, &fec4->key, NULL, pool ) );             /* mr4 should have chained */
+}
+
 int
 main( int argc, char ** argv ) {
   fd_boot( &argc, &argv );
@@ -193,6 +231,7 @@ main( int argc, char ** argv ) {
 
   test_insert( wksp );
   test_publish( wksp );
+  test_slot_mr( 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/repair/fd_repair_tile.c b/src/discof/repair/fd_repair_tile.c
@@ -1108,6 +1108,7 @@ during_housekeeping( fd_repair_tile_ctx_t * ctx ) {
   long now = fd_log_wallclock();
   if( FD_UNLIKELY( now - ctx->tsprint > (long)10e9 ) ) {
     fd_forest_print( ctx->forest );
+    fd_reasm_print( ctx->reasm );
     ctx->tsprint = fd_log_wallclock();
   }
 }
diff --git a/src/discof/replay/fd_replay_tile.c b/src/discof/replay/fd_replay_tile.c
@@ -1447,6 +1447,16 @@ handle_new_slice( fd_replay_tile_ctx_t * ctx, fd_stem_context_t * stem ) {
   ulong slice_sz = 0;
   for( ulong i = 0; i < slice.merkles_cnt; i++ ) {
     fd_store_fec_t * fec = fd_store_query( ctx->store, &slice.merkles[i] );
+    if( FD_UNLIKELY( !fec ) ) {
+
+      /* The only case in which a FEC is not found in the store after
+         repair has notified is if the FEC was on a minority fork that
+         has already been published away.  In this case we abandon the
+         entire slice because it is no longer relevant.  */
+
+      FD_LOG_WARNING(( "store fec for slot: %lu is on minority fork already pruned by publish. abandoning slice. root: %lu. pruned merkle: %s", slice.slot, ctx->root, FD_BASE58_ENC_32_ALLOCA( &slice.merkles[i] ) ));
+      return;
+    }
     FD_TEST( fec );
     memcpy( ctx->slice_exec_ctx.buf + slice_sz, fec->data, fec->data_sz );
     slice_sz += fec->data_sz;

Original file line number	Diff line number	Diff line change
`@@ -1108,6 +1108,7 @@ during_housekeeping( fd_repair_tile_ctx_t * ctx ) {`
`1108`	`1108`	`long now = fd_log_wallclock();`
`1109`	`1109`	`if( FD_UNLIKELY( now - ctx->tsprint > (long)10e9 ) ) {`
`1110`	`1110`	`fd_forest_print( ctx->forest );`
	`1111`	`+ fd_reasm_print( ctx->reasm );`
`1111`	`1112`	`ctx->tsprint = fd_log_wallclock();`
`1112`	`1113`	`}`
`1113`	`1114`	`}`