feat(p2p): expose peer connection metadata

Author: AlisinaDevelo

CHANGELOG.md

@@ -4,6 +4,14 @@ All notable changes to StreamHive are documented here. This project follows [Sem
 
 ## [Unreleased]
 
+### Added
+
+- **`p2p`**: `PeerSnapshots` exposes connected peer metadata for operational tooling.
+
+### Changed
+
+- **Ops**: `/peers` now includes local address, connection timestamp, and connection age.
+
 ## [0.6.0] — 2026-07-02
 
 ### Added

README.md

@@ -50,7 +50,7 @@ Inspect connected peers:
 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.
+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 sorted peer metadata including remote address, local address, direction, connection timestamp, and connection age.
 
 Or run the whole flow:
 
@@ -108,7 +108,7 @@ go run . -listen 127.0.0.1:7070 -replicate -store-dir ./streamhive-data
 
 | Import | Purpose |
 |--------|---------|
-| `github.com/AliSinaDevelo/StreamHive/p2p` | `TCPTransport`, framing (`ReadFrame` / `WriteFrame`), metrics |
+| `github.com/AliSinaDevelo/StreamHive/p2p` | `TCPTransport`, framing (`ReadFrame` / `WriteFrame`), peer snapshots, metrics |
 | `github.com/AliSinaDevelo/StreamHive/replication` | Blob replication messages (`blob.put`, `blob.has`, `blob.get`, `blob.missing`) and store apply helper |
 | `github.com/AliSinaDevelo/StreamHive/storage` | `BlobStore`, `BlobKeyLister`, `MemoryStore`, `FileStore`, SHA-256 content key helpers |
 

docs/ARCHITECTURE.md

@@ -17,7 +17,7 @@ flowchart TD
 
 ## Layers
 
-1. **Transport (`p2p`)** — `TCPTransport` with `context.Context` on `ListenAndAccept` / `Dial`, accept-loop shutdown coordinated with `Close`, optional TLS, optional framed reads via `FrameHandler`, metrics, and peer disconnect hooks.
+1. **Transport (`p2p`)** — `TCPTransport` with `context.Context` on `ListenAndAccept` / `Dial`, accept-loop shutdown coordinated with `Close`, optional TLS, optional framed reads via `FrameHandler`, metrics, peer snapshots, and peer disconnect hooks.
 2. **Framing (`p2p`)** — `SHV1` length-prefixed payloads (`ReadFrame` / `WriteFrame`) with a configurable maximum size (DoS bound). Application-level handshake string: `HandshakeVersionV1`.
 3. **Replication (`replication`)** — typed JSON messages carried inside frames. `blob.put` writes one key/value blob to a receiving `BlobStore`; `blob.has`, `blob.get`, and `blob.missing` provide the inventory/request vocabulary for anti-entropy sync.
 4. **Storage (`storage`)** — `BlobStore` interface with `BlobKeyLister` inventory support, `MemoryStore` for tests/demos, and `FileStore` for durable local blobs; SHA-256 helpers provide stable content-addressed keys.
@@ -137,7 +137,7 @@ 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 `/peers` snapshots for connected peer addresses/direction.
+- JSON `/peers` snapshots for connected peer addresses, direction, connection timestamp, and connection age.
 - JSON `/metrics` counters for stored/sent blobs, bytes, duplicates, and replication errors.
 
 Not implemented yet:

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`, `/peers` (JSON peer snapshot), `/metrics` (JSON counters), `/metrics/prometheus` (Prometheus text).
+- **8080** — HTTP `/livez`, `/readyz`, `/peers` (JSON peer metadata), `/metrics` (JSON counters), `/metrics/prometheus` (Prometheus text).
 
 Use TLS flags (`-tls-cert`, `-tls-key`, `-tls-ca`, …) when exposing services beyond a lab network.
 
@@ -104,5 +104,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.
+- **Peer visibility**: `/peers` returns active connected peers with remote address, local address, direction, connection timestamp, and connection age.
 - **Saturation**: JSON `/metrics` fields `active_peers` and `peers_rejected`, or Prometheus samples from `/metrics/prometheus`.

main.go

@@ -696,21 +696,35 @@ func (m *replicationMetrics) Snapshot() map[string]int64 {
 }
 
 type peerStatus struct {
-	RemoteAddr string `json:"remote_addr"`
-	Outbound   bool   `json:"outbound"`
+	RemoteAddr     string `json:"remote_addr"`
+	LocalAddr      string `json:"local_addr,omitempty"`
+	Outbound       bool   `json:"outbound"`
+	ConnectedAt    string `json:"connected_at,omitempty"`
+	ConnectedForMS int64  `json:"connected_for_ms"`
 }
 
 type peersResponse struct {
 	ActivePeers int          `json:"active_peers"`
 	Peers       []peerStatus `json:"peers"`
 }
 
-func snapshotPeers(peers []p2p.Peer) peersResponse {
+func snapshotPeers(peers []p2p.PeerSnapshot, now time.Time) peersResponse {
 	statuses := make([]peerStatus, 0, len(peers))
 	for _, peer := range peers {
+		connectedAt := ""
+		var connectedForMS int64
+		if !peer.ConnectedAt.IsZero() {
+			connectedAt = peer.ConnectedAt.UTC().Format(time.RFC3339Nano)
+			if now.After(peer.ConnectedAt) {
+				connectedForMS = now.Sub(peer.ConnectedAt).Milliseconds()
+			}
+		}
 		statuses = append(statuses, peerStatus{
-			RemoteAddr: peer.RemoteAddr().String(),
-			Outbound:   peer.IsOutbound(),
+			RemoteAddr:     peer.RemoteAddr,
+			LocalAddr:      peer.LocalAddr,
+			Outbound:       peer.Outbound,
+			ConnectedAt:    connectedAt,
+			ConnectedForMS: connectedForMS,
 		})
 	}
 	sort.Slice(statuses, func(i, j int) bool {
@@ -753,7 +767,7 @@ func startHealth(addr string, tr *p2p.TCPTransport, replMetrics *replicationMetr
 		w.Header().Set("Content-Type", "application/json")
 		enc := json.NewEncoder(w)
 		enc.SetIndent("", "  ")
-		_ = enc.Encode(snapshotPeers(tr.Peers()))
+		_ = enc.Encode(snapshotPeers(tr.PeerSnapshots(), time.Now()))
 	})
 	mux.HandleFunc("/metrics/prometheus", func(w http.ResponseWriter, _ *http.Request) {
 		w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")

main_test.go

@@ -34,20 +34,6 @@ 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()
@@ -630,15 +616,29 @@ func TestWritePrometheusMetrics(t *testing.T) {
 }
 
 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},
-	})
+	connectedAt := time.Date(2026, 7, 2, 1, 2, 3, 4, time.UTC)
+	now := connectedAt.Add(1500 * time.Millisecond)
+	resp := snapshotPeers([]p2p.PeerSnapshot{
+		{RemoteAddr: "127.0.0.1:9002", LocalAddr: "127.0.0.1:7002", Outbound: true, ConnectedAt: connectedAt},
+		{RemoteAddr: "127.0.0.1:9001", LocalAddr: "127.0.0.1:7001", Outbound: false, ConnectedAt: connectedAt},
+	}, now)
 
 	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},
+		{
+			RemoteAddr:     "127.0.0.1:9001",
+			LocalAddr:      "127.0.0.1:7001",
+			Outbound:       false,
+			ConnectedAt:    "2026-07-02T01:02:03.000000004Z",
+			ConnectedForMS: 1500,
+		},
+		{
+			RemoteAddr:     "127.0.0.1:9002",
+			LocalAddr:      "127.0.0.1:7002",
+			Outbound:       true,
+			ConnectedAt:    "2026-07-02T01:02:03.000000004Z",
+			ConnectedForMS: 1500,
+		},
 	}, resp.Peers)
 }
 

p2p/tcp_transport.go

@@ -24,24 +24,31 @@ var ErrTransportClosed = errors.New("p2p: transport closed")
 
 // TCPPeer is a TCP-backed Peer.
 type TCPPeer struct {
-	conn     net.Conn
-	outbound bool
+	conn        net.Conn
+	outbound    bool
+	connectedAt time.Time
 }
 
 // NewTCPPeer wraps a connection as a Peer.
 func NewTCPPeer(conn net.Conn, outbound bool) *TCPPeer {
-	return &TCPPeer{conn: conn, outbound: outbound}
+	return &TCPPeer{conn: conn, outbound: outbound, connectedAt: time.Now().UTC()}
 }
 
 // RemoteAddr returns the remote network address.
 func (p *TCPPeer) RemoteAddr() net.Addr { return p.conn.RemoteAddr() }
 
+// LocalAddr returns the local network address.
+func (p *TCPPeer) LocalAddr() net.Addr { return p.conn.LocalAddr() }
+
 // Close closes the connection.
 func (p *TCPPeer) Close() error { return p.conn.Close() }
 
 // IsOutbound reports whether this peer was created from a dial (outbound).
 func (p *TCPPeer) IsOutbound() bool { return p.outbound }
 
+// ConnectedAt reports when this peer was registered locally.
+func (p *TCPPeer) ConnectedAt() time.Time { return p.connectedAt }
+
 // Conn returns the underlying connection for protocol codecs.
 func (p *TCPPeer) Conn() net.Conn { return p.conn }
 
@@ -52,6 +59,14 @@ func (p *TCPPeer) WriteFrame(payload []byte, maxPayload int) error {
 
 var _ Peer = (*TCPPeer)(nil)
 
+// PeerSnapshot is a point-in-time description of a connected peer.
+type PeerSnapshot struct {
+	RemoteAddr  string
+	LocalAddr   string
+	Outbound    bool
+	ConnectedAt time.Time
+}
+
 // TCPTransport listens on TCP and tracks connected peers.
 type TCPTransport struct {
 	ListenAddress      string
@@ -115,6 +130,25 @@ func (t *TCPTransport) Peers() []Peer {
 	return peers
 }
 
+// PeerSnapshots returns stable metadata for currently connected peers.
+func (t *TCPTransport) PeerSnapshots() []PeerSnapshot {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+	snapshots := make([]PeerSnapshot, 0, len(t.peers))
+	for _, peer := range t.peers {
+		snapshot := PeerSnapshot{
+			RemoteAddr: peer.RemoteAddr().String(),
+			Outbound:   peer.IsOutbound(),
+		}
+		if tcpPeer, ok := peer.(*TCPPeer); ok {
+			snapshot.LocalAddr = tcpPeer.LocalAddr().String()
+			snapshot.ConnectedAt = tcpPeer.ConnectedAt()
+		}
+		snapshots = append(snapshots, snapshot)
+	}
+	return snapshots
+}
+
 func (t *TCPTransport) logger() *slog.Logger {
 	if t.Logger != nil {
 		return t.Logger

p2p/tcp_transport_test.go

@@ -149,6 +149,38 @@ func TestPeersReturnsSnapshot(t *testing.T) {
 	assert.True(t, clientPeers[0].IsOutbound())
 }
 
+func TestPeerSnapshotsIncludesConnectionMetadata(t *testing.T) {
+	ctx := context.Background()
+	before := time.Now().UTC()
+	server := NewTCPTransport("127.0.0.1:0")
+	server.Logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	require.NoError(t, server.ListenAndAccept(ctx))
+	defer func() { _ = server.Close() }()
+
+	client := NewTCPTransport("127.0.0.1:0")
+	client.Logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	require.NoError(t, client.ListenAndAccept(ctx))
+	defer func() { _ = client.Close() }()
+
+	require.NoError(t, client.Dial(ctx, server.Addr().String()))
+	waitFor(t, func() bool {
+		return len(server.PeerSnapshots()) == 1 && len(client.PeerSnapshots()) == 1
+	})
+
+	serverSnapshot := server.PeerSnapshots()[0]
+	clientSnapshot := client.PeerSnapshots()[0]
+	assert.False(t, serverSnapshot.Outbound)
+	assert.True(t, clientSnapshot.Outbound)
+	assert.NotEmpty(t, serverSnapshot.RemoteAddr)
+	assert.NotEmpty(t, serverSnapshot.LocalAddr)
+	assert.NotEmpty(t, clientSnapshot.RemoteAddr)
+	assert.NotEmpty(t, clientSnapshot.LocalAddr)
+	assert.False(t, serverSnapshot.ConnectedAt.Before(before))
+	assert.False(t, clientSnapshot.ConnectedAt.Before(before))
+	assert.False(t, serverSnapshot.ConnectedAt.After(time.Now().UTC()))
+	assert.False(t, clientSnapshot.ConnectedAt.After(time.Now().UTC()))
+}
+
 func TestDial_contextCancelled(t *testing.T) {
 	ctx := context.Background()
 	server := NewTCPTransport("127.0.0.1:0")