unified labels and name consistancy

Author: ihlec

internal/managers/shard/shard_merge.go

@@ -14,7 +14,11 @@ import (
 	"dlockss/pkg/schema"
 )
 
-// writeJSONError writes a JSON {"error":"..."} response with the given status code.
+func writeJSON(w http.ResponseWriter, v interface{}) {
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(v)
+}
+
 func writeJSONError(w http.ResponseWriter, msg string, code int) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(code)
@@ -127,7 +131,7 @@ func (m *Monitor) handleHeartbeatWithRole(ctx context.Context, senderID peer.ID,
 		if nodeName != "" {
 			logName = nodeName + " (" + peerIDStr + ")"
 		}
-		slog.Info("new node discovered via heartbeat", "peer", logName, "shard", shardLogLabel(shardID), "pinned", pinnedCount, "role", role)
+		slog.Info("new node discovered via heartbeat", "peer", logName, "shard", shardLabel(shardID), "pinned", pinnedCount, "role", role)
 		nodeState = &NodeState{
 			PeerID:         peerIDStr,
 			NodeName:       nodeName,
@@ -158,9 +162,7 @@ func (m *Monitor) handleHeartbeatWithRole(ctx context.Context, senderID peer.ID,
 			}
 			// Ignore pinned=0 during grace period: stale heartbeats can arrive
 			// before the node finishes its first pin cycle.
-			if now.Sub(firstSeen) < unpinGracePeriod {
-				// nop
-			} else {
+			if now.Sub(firstSeen) >= unpinGracePeriod {
 				removedFromManifests := 0
 				for manifest, peers := range m.manifestReplication {
 					if _, had := peers[peerIDStr]; had {
@@ -172,7 +174,7 @@ func (m *Monitor) handleHeartbeatWithRole(ctx context.Context, senderID peer.ID,
 					}
 				}
 				if removedFromManifests > 0 {
-					slog.Info("unpin all", "peer", peerIDStr, "shard", shardLogLabel(shardID), "removed_manifests", removedFromManifests)
+					slog.Info("unpin all", "peer", peerIDStr, "shard", shardLabel(shardID), "removed_manifests", removedFromManifests)
 				}
 			}
 		} else {
@@ -246,12 +248,12 @@ func (m *Monitor) updateNodeShardLocked(node *NodeState, newShard string, timest
 	}
 
 	m.treeDirty = true
-	slog.Info("shard move", "peer", node.PeerID, "from", shardLogLabel(lastShard), "to", shardLogLabel(newShard))
+	slog.Info("shard move", "peer", node.PeerID, "from", shardLabel(lastShard), "to", shardLabel(newShard))
 
 	isSplit := len(newShard) > len(lastShard) && strings.HasPrefix(newShard, lastShard)
 	if isSplit {
 		if !m.hasSplitEvent(lastShard, newShard) {
-			slog.Info("detected shard split", "parent", shardLogLabel(lastShard), "child", newShard, "peer", node.PeerID)
+			slog.Info("detected shard split", "parent", shardLabel(lastShard), "child", newShard, "peer", node.PeerID)
 		}
 		m.lastSplitTime = timestamp
 		m.splitEvents = append(m.splitEvents, ShardSplitEvent{
@@ -298,7 +300,7 @@ func (m *Monitor) updateNodeShardLocked(node *NodeState, newShard string, timest
 		}
 	}
 	if removed > 0 {
-		slog.Info("shard move removed peer from manifests", "peer", peerIDStr, "removed_manifests", removed, "shard", shardLogLabel(newShard))
+		slog.Info("shard move removed peer from manifests", "peer", peerIDStr, "removed_manifests", removed, "shard", shardLabel(newShard))
 	}
 	if isSiblingShard(lastShard, newShard) {
 		m.peerLastSiblingMove[peerIDStr] = siblingMoveRecord{from: lastShard, to: newShard, when: timestamp}

internal/monitor/monitor_handlers.go

@@ -143,7 +143,7 @@ func (n *NodeState) EffectiveShard() string {
 	return ""
 }
 
-func shardLogLabel(shardID string) string {
+func shardLabel(shardID string) string {
 	if shardID == "" {
 		return "root"
 	}

internal/monitor/monitor_models.go

@@ -134,8 +134,6 @@ func (mr manifestResult) isAtTarget() bool {
 	return mr.count >= minRep && mr.count <= mr.maxRep
 }
 
-// --- Public methods using the shared helpers ---
-
 func (m *Monitor) runReplicationCleanup() {
 	ticker := time.NewTicker(ReplicationCleanupEvery)
 	defer ticker.Stop()
@@ -147,15 +145,15 @@ func (m *Monitor) runReplicationCleanup() {
 		}
 		m.mu.Lock()
 		cutoff := time.Now().Add(-ReplicationAnnounceTTL)
-		for manifest, peers := range m.manifestReplication {
+		for cid, peers := range m.manifestReplication {
 			for peerID, lastSeen := range peers {
 				if lastSeen.Before(cutoff) {
 					delete(peers, peerID)
 				}
 			}
 			if len(peers) == 0 {
-				delete(m.manifestReplication, manifest)
-				delete(m.manifestShard, manifest)
+				delete(m.manifestReplication, cid)
+				delete(m.manifestShard, cid)
 			}
 		}
 		m.mu.Unlock()
@@ -178,11 +176,7 @@ func (m *Monitor) runReplicationCleanup() {
 		for _, shard := range shardLabels {
 			peers := membership[shard]
 			atTarget := byShard[shard]
-			shardLabel := shard
-			if shardLabel == "" {
-				shardLabel = "(root)"
-			}
-			fmt.Fprintf(&b, " %s: %d nodes [%s] %d files at target;", shardLabel, len(peers), strings.Join(peers, ","), atTarget)
+			fmt.Fprintf(&b, " %s: %d nodes [%s] %d files at target;", shardLabel(shard), len(peers), strings.Join(peers, ","), atTarget)
 		}
 		slog.Info("replication snapshot", "total_nodes", totalNodes, "total_manifests", totalFiles, "total_at_target", filesAtTarget, "avg_replication", fmt.Sprintf("%.2f", avgLevel), "shards", strings.TrimSpace(b.String()))
 		if filesAtTarget == 0 && totalFiles > 0 && totalNodes > 0 {
@@ -277,12 +271,12 @@ func (m *Monitor) getReplicationStats() (distribution [11]int, avgLevel float64,
 	rs := m.newReplicationSnapshotUnlocked()
 	var totalReplication, manifestCount int
 
-	for manifest, peers := range m.manifestReplication {
+	for cid, peers := range m.manifestReplication {
 		if len(peers) == 0 {
 			continue
 		}
 		shardCounts := rs.buildShardCounts(peers)
-		mr := rs.resolveManifest(manifest, peers, shardCounts)
+		mr := rs.resolveManifest(cid, peers, shardCounts)
 		if mr.count == 0 {
 			continue
 		}
@@ -313,22 +307,18 @@ func (m *Monitor) getReplicationCIDsByLevel(level int) []CIDEntry {
 	rs := m.newReplicationSnapshotUnlocked()
 	var result []CIDEntry
 
-	for manifest, peers := range m.manifestReplication {
+	for cid, peers := range m.manifestReplication {
 		if len(peers) == 0 {
 			continue
 		}
 		shardCounts := rs.buildShardCounts(peers)
-		mr := rs.resolveManifest(manifest, peers, shardCounts)
+		mr := rs.resolveManifest(cid, peers, shardCounts)
 		if mr.count == 0 {
 			continue
 		}
 		matches := (level == 10 && mr.count >= 10) || (level < 10 && mr.count == level)
 		if matches {
-			shardLabel := m.manifestShard[manifest]
-			if shardLabel == "" {
-				shardLabel = "root"
-			}
-			result = append(result, CIDEntry{CID: manifest, Shard: shardLabel, Replicas: mr.count})
+			result = append(result, CIDEntry{CID: cid, Shard: shardLabel(m.manifestShard[cid]), Replicas: mr.count})
 		}
 	}
 	sort.Slice(result, func(i, j int) bool { return result[i].CID < result[j].CID })
@@ -347,19 +337,19 @@ func (m *Monitor) getReplicationByShard() map[string]int {
 		peers       map[string]time.Time
 	}
 	manifests := make(map[string]manifestInfo, len(m.manifestReplication))
-	for manifest, peers := range m.manifestReplication {
+	for cid, peers := range m.manifestReplication {
 		if len(peers) == 0 {
 			continue
 		}
-		manifests[manifest] = manifestInfo{
+		manifests[cid] = manifestInfo{
 			shardCounts: rs.buildShardCounts(peers),
 			peers:       peers,
 		}
 	}
 
 	filesAtTargetPerShard := make(map[string]int)
-	for manifest, info := range manifests {
-		mr := rs.resolveManifest(manifest, info.peers, info.shardCounts)
+	for cid, info := range manifests {
+		mr := rs.resolveManifest(cid, info.peers, info.shardCounts)
 		if mr.count == 0 || mr.maxRep == 0 {
 			continue
 		}

internal/monitor/monitor_replication.go

@@ -132,14 +132,12 @@ func (m *Monitor) handleNodes(w http.ResponseWriter, r *http.Request) {
 			strings.Contains(strings.ToLower(node.NodeName), query)
 	})
 	response := m.buildNodeResponse(snapshot)
-	w.Header().Set("Content-Type", "application/json")
-	json.NewEncoder(w).Encode(response)
+	writeJSON(w, response)
 }
 
 func (m *Monitor) handleShardTree(w http.ResponseWriter, r *http.Request) {
 	tree := m.GetShardTree()
-	w.Header().Set("Content-Type", "application/json")
-	json.NewEncoder(w).Encode(tree)
+	writeJSON(w, tree)
 }
 
 func (m *Monitor) handleShardNodes(w http.ResponseWriter, r *http.Request) {
@@ -148,16 +146,10 @@ func (m *Monitor) handleShardNodes(w http.ResponseWriter, r *http.Request) {
 		return shard == shardFilter
 	})
 	response := m.buildNodeResponse(snapshot)
-	shardLabel := shardFilter
-	if shardLabel == "" {
-		shardLabel = "root"
-	}
-	w.Header().Set("Content-Type", "application/json")
-	json.NewEncoder(w).Encode(map[string]interface{}{"shard_id": shardFilter, "shard_label": shardLabel, "nodes": response, "count": len(response)})
+	writeJSON(w, map[string]interface{}{"shard_id": shardFilter, "shard_label": shardLabel(shardFilter), "nodes": response, "count": len(response)})
 }
 
 func (m *Monitor) handleRootTopic(w http.ResponseWriter, r *http.Request) {
-	w.Header().Set("Content-Type", "application/json")
 	if r.Method == http.MethodPost {
 		var body struct {
 			TopicPrefix string `json:"topic_prefix,omitempty"`
@@ -183,7 +175,7 @@ func (m *Monitor) writeRootTopicResponse(w http.ResponseWriter) {
 	prefix := m.getTopicPrefix()
 	topic := m.getTopicName()
 	rootTopic := fmt.Sprintf("%s-%s-shard-", prefix, topic)
-	json.NewEncoder(w).Encode(map[string]string{
+	writeJSON(w, map[string]string{
 		"root_topic":   rootTopic,
 		"topic_prefix": prefix,
 		"topic_name":   topic,
@@ -204,16 +196,14 @@ func (m *Monitor) handleNodeFiles(w http.ResponseWriter, r *http.Request) {
 		entries = []CIDEntry{}
 	}
 	m.mu.RUnlock()
-	w.Header().Set("Content-Type", "application/json")
-	json.NewEncoder(w).Encode(map[string]interface{}{"peer_id": peerID, "cids": entries, "count": len(entries)})
+	writeJSON(w, map[string]interface{}{"peer_id": peerID, "cids": entries, "count": len(entries)})
 }
 
 func (m *Monitor) handleUniqueCIDs(w http.ResponseWriter, r *http.Request) {
 	m.mu.RLock()
 	entries := m.buildCIDEntriesUnlocked(m.uniqueCIDs)
 	m.mu.RUnlock()
-	w.Header().Set("Content-Type", "application/json")
-	json.NewEncoder(w).Encode(map[string]interface{}{"cids": entries, "count": len(entries)})
+	writeJSON(w, map[string]interface{}{"cids": entries, "count": len(entries)})
 }
 
 func (m *Monitor) handleReplication(w http.ResponseWriter, r *http.Request) {
@@ -223,9 +213,8 @@ func (m *Monitor) handleReplication(w http.ResponseWriter, r *http.Request) {
 	}
 	dist, avg, atTarget := m.getReplicationStats()
 	byShard := m.getReplicationByShard()
-	w.Header().Set("Content-Type", "application/json")
 	w.Header().Set("Cache-Control", "no-store, no-cache, must-revalidate")
-	json.NewEncoder(w).Encode(map[string]interface{}{
+	writeJSON(w, map[string]interface{}{
 		"replication_distribution":  dist,
 		"avg_replication_level":     avg,
 		"files_at_target":           atTarget,
@@ -246,8 +235,7 @@ func (m *Monitor) handleReplicationCIDs(w http.ResponseWriter, r *http.Request)
 		return
 	}
 	entries := m.getReplicationCIDsByLevel(level)
-	w.Header().Set("Content-Type", "application/json")
-	json.NewEncoder(w).Encode(map[string]interface{}{"level": level, "cids": entries, "count": len(entries)})
+	writeJSON(w, map[string]interface{}{"level": level, "cids": entries, "count": len(entries)})
 }
 
 func (m *Monitor) handleManifestPayload(w http.ResponseWriter, r *http.Request) {
@@ -284,17 +272,16 @@ func (m *Monitor) handleManifestPayload(w http.ResponseWriter, r *http.Request)
 		"size":        ro.TotalSize,
 		"sig":         base64.StdEncoding.EncodeToString(ro.Signature),
 	}
-	w.Header().Set("Content-Type", "application/json")
-	json.NewEncoder(w).Encode(map[string]interface{}{"payload_cid": ro.Payload.String(), "manifest": manifest})
+	writeJSON(w, map[string]interface{}{"payload_cid": ro.Payload.String(), "manifest": manifest})
 }
 
 func (m *Monitor) handleIdentify(w http.ResponseWriter, r *http.Request) {
-	peerStr := strings.TrimSpace(r.URL.Query().Get("peer"))
-	if peerStr == "" {
+	peerIDStr := strings.TrimSpace(r.URL.Query().Get("peer"))
+	if peerIDStr == "" {
 		writeJSONError(w, "missing peer parameter", http.StatusBadRequest)
 		return
 	}
-	pid, err := peer.Decode(peerStr)
+	pid, err := peer.Decode(peerIDStr)
 	if err != nil {
 		writeJSONError(w, "invalid peer ID", http.StatusBadRequest)
 		return
@@ -335,8 +322,7 @@ func (m *Monitor) handleIdentify(w http.ResponseWriter, r *http.Request) {
 		"connected":        connected,
 		"region":           region,
 	}
-	w.Header().Set("Content-Type", "application/json")
-	json.NewEncoder(w).Encode(result)
+	writeJSON(w, result)
 }
 
 func (m *Monitor) handleDashboard(w http.ResponseWriter, r *http.Request) {
@@ -376,7 +362,7 @@ func (m *Monitor) RunStatusLogger(ctx context.Context) {
 			sort.Strings(shardIDs)
 			parts := make([]string, 0, len(shardIDs))
 			for _, sid := range shardIDs {
-				parts = append(parts, fmt.Sprintf("%s: %d", shardLogLabel(sid), shardCounts[sid]))
+				parts = append(parts, fmt.Sprintf("%s: %d", shardLabel(sid), shardCounts[sid]))
 			}
 			slog.Info("status", "nodes", nodeCount, "shards", len(shardCounts), "pinned", totalPinned, "detail", strings.Join(parts, ", "))
 		}

internal/monitor/monitor_routes.go

@@ -96,7 +96,7 @@ func (m *Monitor) ensureShardSubscriptionUnlocked(ctx context.Context, shardID s
 		_ = topic.Publish(ctx, probeMsg)
 	}
 	go m.handleShardMessages(m.subCtx, sub, shardID, topicName)
-	slog.Info("subscribed to shard topic", "shard", shardLogLabel(shardID))
+	slog.Info("subscribed to shard topic", "shard", shardLabel(shardID))
 }
 
 // resolveIPFromPeer extracts the best public IP address for the given peer
@@ -169,7 +169,7 @@ func (m *Monitor) handleShardMessages(ctx context.Context, sub *pubsub.Subscript
 			if peerStr := strings.TrimSpace(string(data[6:])); peerStr != "" {
 				if leaveID, err := decodePeerIDWithFallback(peerStr); err == nil {
 					m.handleLeaveShard(ctx, leaveID, shardID)
-					slog.Info("shard leave", "peer", leaveID.String(), "shard", shardLogLabel(shardID))
+					slog.Info("shard leave", "peer", leaveID.String(), "shard", shardLabel(shardID))
 				}
 			}
 
@@ -263,7 +263,7 @@ func (m *Monitor) dispatchJoin(ctx context.Context, data []byte, senderID peer.I
 		if nodeName != "" {
 			logLabel = nodeName + " (" + joinID.String() + ")"
 		}
-		slog.Info("shard join", "peer", logLabel, "shard", shardLogLabel(shardID), "role", role)
+		slog.Info("shard join", "peer", logLabel, "shard", shardLabel(shardID), "role", role)
 	}
 }