Self Healing

Author: j-rafique

p2p/kademlia/dht.go

@@ -210,12 +210,12 @@ func (s *DHT) eligibleForRouting(n *Node) bool {
 	if integrationTestEnabled() {
 		return true
 	}
-	// Bootstrap-safe behavior: until first non-empty chain allowlist arrives,
-	// keep routing/read gating disabled to avoid accidental lockout.
+	// Bootstrap fail-open: until the first successful allowlist sync, do not block routing.
+	// This avoids startup isolation during transient chain/API failures.
 	if !s.routingAllowReady.Load() {
 		return true
 	}
-	// Once initialized, an empty active set means no routing-eligible peers.
+	// Once initialized, an empty routing set means no routing-eligible peers.
 	if s.routingAllowCount.Load() == 0 {
 		return false
 	}
@@ -274,8 +274,8 @@ func (s *DHT) filterEligibleNodes(nodes []*Node) []*Node {
 	if integrationTestEnabled() {
 		return nodes
 	}
-	// If the routing allowlist has not been initialized yet, keep gating disabled
-	// but still sanitize malformed node entries.
+	// Bootstrap fail-open: before first allowlist sync, keep routing candidates
+	// but sanitize malformed node entries.
 	if !s.routingAllowReady.Load() {
 		out := nodes[:0]
 		for _, n := range nodes {
@@ -286,7 +286,7 @@ func (s *DHT) filterEligibleNodes(nodes []*Node) []*Node {
 		}
 		return out
 	}
-	// Once initialized, empty means no routing-eligible peers.
+	// Once initialized, an empty allowlist means no routing-eligible peers.
 	if s.routingAllowCount.Load() == 0 {
 		return nil
 	}
@@ -514,6 +514,7 @@ func (s *DHT) Start(ctx context.Context) error {
 	}
 
 	go s.StartReplicationWorker(ctx)
+	go s.startRebalanceWorker(ctx)
 	go s.startDisabledKeysCleanupWorker(ctx)
 	// TEMPORARY: disabled to pause redundant-key classification into del_keys.
 	// Re-enable once deletion-safety behavior is finalized.
@@ -2099,7 +2100,7 @@ func (s *DHT) addNode(ctx context.Context, node *Node) *Node {
 	node.SetHashedID()
 
 	// Chain-state routing gate (enabled after allowlist initialization):
-	// only chain-allowlisted peers may enter the routing table.
+	// only routing-eligible supernodes (Active + Postponed) can be admitted.
 	if !s.eligibleForRouting(node) {
 		logtrace.Debug(ctx, "Rejecting node: not eligible for routing", logtrace.Fields{
 			logtrace.FieldModule: "p2p",

p2p/kademlia/message.go

@@ -30,6 +30,8 @@ const (
 	BatchFindNode
 	// BatchGetValues finds values in kademlia network
 	BatchGetValues
+	// BatchProbeKeys checks local key presence/status without side effects
+	BatchProbeKeys
 )
 
 func init() {
@@ -49,6 +51,8 @@ func init() {
 	gob.Register(&BatchFindNodeResponse{})
 	gob.Register(&BatchGetValuesRequest{})
 	gob.Register(&BatchGetValuesResponse{})
+	gob.Register(&BatchProbeKeysRequest{})
+	gob.Register(&BatchProbeKeysResponse{})
 }
 
 type MessageWithError struct {
@@ -166,6 +170,17 @@ type BatchGetValuesResponse struct {
 	Status ResponseStatus
 }
 
+// BatchProbeKeysRequest defines the request data for side-effect-free local key probes.
+type BatchProbeKeysRequest struct {
+	Keys []string // hex-encoded keys
+}
+
+// BatchProbeKeysResponse defines the response data for local key probes.
+type BatchProbeKeysResponse struct {
+	Status ResponseStatus
+	Data   map[string]LocalKeyStatus // key -> local status
+}
+
 // encode the message
 func encode(message *Message) ([]byte, error) {
 	var buf bytes.Buffer

p2p/kademlia/network.go

@@ -47,6 +47,7 @@ func init() {
 		FindValue:       5 * time.Second,
 		BatchFindValues: 60 * time.Second, // responder compresses
 		BatchGetValues:  75 * time.Second, // large, sometimes cloud fetch then send back
+		BatchProbeKeys:  10 * time.Second,
 		StoreData:       10 * time.Second,
 		BatchStoreData:  75 * time.Second, // large uncompressed payloads
 		Replicate:       90 * time.Second,
@@ -235,6 +236,37 @@ func (s *Network) handleFindValue(ctx context.Context, message *Message) (res []
 	return s.encodeMesage(resMsg)
 }
 
+func (s *Network) handleBatchProbeKeys(ctx context.Context, message *Message) (res []byte, err error) {
+	defer func() {
+		if response, err := s.handlePanic(ctx, message.Sender, BatchProbeKeys); response != nil || err != nil {
+			res = response
+		}
+	}()
+
+	request, ok := message.Data.(*BatchProbeKeysRequest)
+	if !ok {
+		err := errors.New("invalid BatchProbeKeysRequest")
+		return s.generateResponseMessage(ctx, BatchProbeKeys, message.Sender, ResultFailed, err.Error())
+	}
+
+	// Keep routing table fresh while handling probe traffic.
+	s.dht.addNode(ctx, message.Sender)
+
+	statuses, err := s.dht.store.RetrieveBatchLocalStatus(ctx, request.Keys)
+	if err != nil {
+		err = errors.Errorf("batch probe keys: %w", err)
+		return s.generateResponseMessage(ctx, BatchProbeKeys, message.Sender, ResultFailed, err.Error())
+	}
+
+	response := &BatchProbeKeysResponse{
+		Status: ResponseStatus{Result: ResultOk},
+		Data:   statuses,
+	}
+
+	resMsg := s.dht.newMessage(BatchProbeKeys, message.Sender, response)
+	return s.encodeMesage(resMsg)
+}
+
 func (s *Network) handleStoreData(ctx context.Context, message *Message) (res []byte, err error) {
 	defer func() {
 		if response, err := s.handlePanic(ctx, message.Sender, StoreData); response != nil || err != nil {
@@ -480,6 +512,10 @@ func (s *Network) handleConn(ctx context.Context, rawConn net.Conn) {
 			response, hErr = s.withMetrics(BatchGetValues, func() ([]byte, error) {
 				return s.handleGetValuesRequest(ctx, request, reqID)
 			})
+		case BatchProbeKeys:
+			response, hErr = s.withMetrics(BatchProbeKeys, func() ([]byte, error) {
+				return s.handleBatchProbeKeys(ctx, request)
+			})
 		default:
 			// count unknown types as failure and return
 			m := s.metricsFor(mt)
@@ -1321,6 +1357,8 @@ func (s *Network) generateResponseMessage(ctx context.Context, messageType int,
 		response = &ReplicateDataResponse{Status: responseStatus}
 	case BatchGetValues:
 		response = &BatchGetValuesResponse{Status: responseStatus}
+	case BatchProbeKeys:
+		response = &BatchProbeKeysResponse{Status: responseStatus}
 	default:
 		return nil, fmt.Errorf("unsupported message type %d", messageType)
 	}
@@ -1461,6 +1499,8 @@ func msgName(t int) string {
 		return "BatchStoreData"
 	case Replicate:
 		return "Replicate"
+	case BatchProbeKeys:
+		return "BatchProbeKeys"
 	default:
 		return fmt.Sprintf("Type_%d", t)
 	}
@@ -1501,7 +1541,7 @@ func (s *Network) HandleMetricsSnapshot() map[string]HandleCounters {
 func readDeadlineFor(msgType int, overall time.Duration) time.Duration {
 	small := 10 * time.Second
 	switch msgType {
-	case Ping, FindNode, BatchFindNode, FindValue, StoreData, BatchStoreData:
+	case Ping, FindNode, BatchFindNode, FindValue, StoreData, BatchStoreData, BatchProbeKeys:
 		if overall > small+1*time.Second {
 			return small
 		}