feat(ops): expose peer snapshot endpoint

Author: AlisinaDevelo

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to StreamHive are documented here. This project follows [Sem
 
 ## [Unreleased]
 
+### Added
+
+- **Ops**: `/peers` JSON endpoint for inspecting active peer addresses and connection direction.
+
 ## [0.5.0] — 2026-07-01
 
 ### Added

README.md

@@ -2,7 +2,7 @@
 
 [![CI](https://github.com/AliSinaDevelo/StreamHive/actions/workflows/ci.yml/badge.svg)](https://github.com/AliSinaDevelo/StreamHive/actions/workflows/ci.yml)
 
-StreamHive is a **Go library and CLI** for experimenting with distributed, content-addressed storage. It ships a production-minded **TCP transport** (context-aware listen/dial, TLS hooks, framing, metrics, limits), a **length-prefixed wire format** (`SHV1`), a typed **blob replication protocol**, memory and file-backed **blob stores**, and operational endpoints (`/livez`, `/readyz`, `/metrics`, `/metrics/prometheus`).
+StreamHive is a **Go library and CLI** for experimenting with distributed, content-addressed storage. It ships a production-minded **TCP transport** (context-aware listen/dial, TLS hooks, framing, metrics, limits), a **length-prefixed wire format** (`SHV1`), a typed **blob replication protocol**, memory and file-backed **blob stores**, and operational endpoints (`/livez`, `/readyz`, `/peers`, `/metrics`, `/metrics/prometheus`).
 
 **Semver:** public API versions are tracked in [CHANGELOG.md](CHANGELOG.md) and [internal/version/version.go](internal/version/version.go) (currently **v0.5.0**, pre-1.0).
 
@@ -21,7 +21,7 @@ go test ./...
 go run . -version
 make run
 ./bin/fs -listen :7070 -dial 127.0.0.1:8080
-./bin/fs -listen 127.0.0.1:0 -health 127.0.0.1:8080   # HTTP live/ready/metrics
+./bin/fs -listen 127.0.0.1:0 -health 127.0.0.1:8080   # HTTP live/ready/peers/metrics
 ```
 
 ### Two-node replication demo
@@ -44,7 +44,13 @@ Inspect counters:
 curl -s http://127.0.0.1:8080/metrics
 ```
 
-Look for `replication_blobs_stored`, `replication_bytes_stored`, duplicate counters, and transport frame counters. The sender derives the blob key from `SHA-256(put-data)` when `-put-content-key` is set; receivers verify SHA-256-shaped keys before storing. Use `/metrics` for JSON or `/metrics/prometheus` for Prometheus text format.
+Inspect connected peers:
+
+```bash
+curl -s http://127.0.0.1:8080/peers
+```
+
+Look for `replication_blobs_stored`, `replication_bytes_stored`, duplicate counters, and transport frame counters. The sender derives the blob key from `SHA-256(put-data)` when `-put-content-key` is set; receivers verify SHA-256-shaped keys before storing. Use `/metrics` for JSON counters, `/metrics/prometheus` for Prometheus text format, and `/peers` for a sorted peer snapshot.
 
 Or run the whole flow:
 
@@ -106,7 +112,7 @@ Wire handshake string constant: `p2p.HandshakeVersionV1` (carry inside applicati
 | `-peer-reconnect` | Retry `-peers` with exponential backoff |
 | `-peer-reconnect-min` / `-peer-reconnect-max` | Reconnect backoff bounds |
 | `-sync-interval` | Periodically advertise local blob keys to connected peers (0 = startup only) |
-| `-health` | HTTP `host:port` for `/livez`, `/readyz`, `/metrics` |
+| `-health` | HTTP `host:port` for `/livez`, `/readyz`, `/peers`, `/metrics` |
 | `-max-peers` | Cap simultaneous peers (0 = unlimited) |
 | `-dial-timeout` | Outbound dial timeout |
 | `-read-idle-timeout` | Peer read deadline refresh |

docs/ARCHITECTURE.md

@@ -12,7 +12,7 @@ flowchart TD
   rep --> st["Storage (storage)<br/>BlobStore"]
   st --> mem["MemoryStore"]
   st --> file["FileStore (durable)"]
-  tr -. exposes .-> health["HTTP /livez /readyz /metrics"]
+  tr -. exposes .-> health["HTTP /livez /readyz /peers /metrics"]
 ```
 
 ## Layers
@@ -137,11 +137,12 @@ Implemented:
 - Message types: `blob.put`, `blob.has`, `blob.get`, and `blob.missing`.
 - Startup anti-entropy for connected `-replicate` peers.
 - Receiver-side storage via `storage.MemoryStore` or durable `storage.FileStore` with `-store-dir`.
-- JSON `/metrics` counters for stored/sent blobs, bytes, and replication errors.
+- JSON `/peers` snapshots for connected peer addresses/direction.
+- JSON `/metrics` counters for stored/sent blobs, bytes, duplicates, and replication errors.
 
 Not implemented yet:
 
-- Continuous anti-entropy, retries for partial sync failures, or conflict resolution.
+- Retries for partial sync failures or conflict resolution.
 - Automated peer discovery beyond static dial targets.
 - Authenticated application-level identity beyond optional TLS/mTLS configuration.
 

docs/DEPLOYMENT.md

@@ -12,7 +12,7 @@ docker run --rm -p 7070:7070 -p 8080:8080 streamhive:local \
 ```
 
 - **7070** — P2P TCP listener (example).
-- **8080** — HTTP `/livez`, `/readyz`, `/metrics` (JSON counters), `/metrics/prometheus` (Prometheus text).
+- **8080** — HTTP `/livez`, `/readyz`, `/peers` (JSON peer snapshot), `/metrics` (JSON counters), `/metrics/prometheus` (Prometheus text).
 
 Use TLS flags (`-tls-cert`, `-tls-key`, `-tls-ca`, …) when exposing services beyond a lab network.
 
@@ -88,4 +88,5 @@ Define error budgets once you expose a workload to users. Baseline probes:
 
 - **Availability**: `/livez` success rate.
 - **Readiness**: `/readyz` reflects listener bound (`TCPTransport.Ready`).
+- **Peer visibility**: `/peers` returns active connected peers and whether each connection is outbound.
 - **Saturation**: JSON `/metrics` fields `active_peers` and `peers_rejected`, or Prometheus samples from `/metrics/prometheus`.

main.go

@@ -50,7 +50,7 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) error {
 	peerReconnectMin := fs.Duration("peer-reconnect-min", 500*time.Millisecond, "minimum reconnect backoff for -peer-reconnect")
 	peerReconnectMax := fs.Duration("peer-reconnect-max", 30*time.Second, "maximum reconnect backoff for -peer-reconnect")
 	syncInterval := fs.Duration("sync-interval", 0, "periodically advertise local blob keys to connected peers (0 = startup only)")
-	health := fs.String("health", "", "optional HTTP listen addr for /livez /readyz /metrics (e.g. :8080)")
+	health := fs.String("health", "", "optional HTTP listen addr for /livez /readyz /peers /metrics (e.g. :8080)")
 	maxPeers := fs.Int("max-peers", 0, "max simultaneous peers (0 = unlimited)")
 	dialTimeout := fs.Duration("dial-timeout", 0, "default dial timeout (0 = use context only)")
 	readIdle := fs.Duration("read-idle-timeout", 0, "TCP read deadline refresh for peer loops (0 = none for discard mode)")
@@ -695,6 +695,36 @@ func (m *replicationMetrics) Snapshot() map[string]int64 {
 	}
 }
 
+type peerStatus struct {
+	RemoteAddr string `json:"remote_addr"`
+	Outbound   bool   `json:"outbound"`
+}
+
+type peersResponse struct {
+	ActivePeers int          `json:"active_peers"`
+	Peers       []peerStatus `json:"peers"`
+}
+
+func snapshotPeers(peers []p2p.Peer) peersResponse {
+	statuses := make([]peerStatus, 0, len(peers))
+	for _, peer := range peers {
+		statuses = append(statuses, peerStatus{
+			RemoteAddr: peer.RemoteAddr().String(),
+			Outbound:   peer.IsOutbound(),
+		})
+	}
+	sort.Slice(statuses, func(i, j int) bool {
+		if statuses[i].RemoteAddr == statuses[j].RemoteAddr {
+			return !statuses[i].Outbound && statuses[j].Outbound
+		}
+		return statuses[i].RemoteAddr < statuses[j].RemoteAddr
+	})
+	return peersResponse{
+		ActivePeers: len(statuses),
+		Peers:       statuses,
+	}
+}
+
 func startHealth(addr string, tr *p2p.TCPTransport, replMetrics *replicationMetrics, log *slog.Logger) (*http.Server, error) {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/livez", func(w http.ResponseWriter, _ *http.Request) {
@@ -719,6 +749,12 @@ func startHealth(addr string, tr *p2p.TCPTransport, replMetrics *replicationMetr
 		enc.SetIndent("", "  ")
 		_ = enc.Encode(snapshot)
 	})
+	mux.HandleFunc("/peers", func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		enc := json.NewEncoder(w)
+		enc.SetIndent("", "  ")
+		_ = enc.Encode(snapshotPeers(tr.Peers()))
+	})
 	mux.HandleFunc("/metrics/prometheus", func(w http.ResponseWriter, _ *http.Request) {
 		w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
 		snapshot := tr.Metrics().Snapshot()

main_test.go

@@ -34,6 +34,20 @@ func (testPeer) RemoteAddr() net.Addr { return &net.TCPAddr{IP: net.IPv4(127, 0,
 func (testPeer) Close() error         { return nil }
 func (testPeer) IsOutbound() bool     { return false }
 
+type staticPeer struct {
+	addr     string
+	outbound bool
+}
+
+func (p staticPeer) RemoteAddr() net.Addr { return staticAddr(p.addr) }
+func (p staticPeer) Close() error         { return nil }
+func (p staticPeer) IsOutbound() bool     { return p.outbound }
+
+type staticAddr string
+
+func (a staticAddr) Network() string { return "tcp" }
+func (a staticAddr) String() string  { return string(a) }
+
 func (s *safeBuffer) Write(p []byte) (int, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -122,6 +136,15 @@ func TestRun_healthEndpoints(t *testing.T) {
 	assert.Contains(t, string(body), "streamhive_active_peers")
 	assert.Contains(t, string(body), "streamhive_replication_blobs_stored")
 
+	resp5, err := client.Get(base + "/peers")
+	require.NoError(t, err)
+	defer func() { _ = resp5.Body.Close() }()
+	assert.Equal(t, http.StatusOK, resp5.StatusCode)
+	var peers peersResponse
+	require.NoError(t, json.NewDecoder(resp5.Body).Decode(&peers))
+	assert.Equal(t, 0, peers.ActivePeers)
+	assert.Empty(t, peers.Peers)
+
 	cancel()
 	<-errCh
 }
@@ -606,6 +629,19 @@ func TestWritePrometheusMetrics(t *testing.T) {
 	assert.Equal(t, "streamhive_a_metric 1\nstreamhive_z_metric 2\n", out.String())
 }
 
+func TestSnapshotPeersSortsByAddress(t *testing.T) {
+	resp := snapshotPeers([]p2p.Peer{
+		staticPeer{addr: "127.0.0.1:9002", outbound: true},
+		staticPeer{addr: "127.0.0.1:9001", outbound: false},
+	})
+
+	require.Equal(t, 2, resp.ActivePeers)
+	require.Equal(t, []peerStatus{
+		{RemoteAddr: "127.0.0.1:9001", Outbound: false},
+		{RemoteAddr: "127.0.0.1:9002", Outbound: true},
+	}, resp.Peers)
+}
+
 func TestValidateReconnectBackoff(t *testing.T) {
 	assert.NoError(t, validateReconnectBackoff(10*time.Millisecond, 20*time.Millisecond))
 	assert.Error(t, validateReconnectBackoff(0, 20*time.Millisecond))