feat: add content-addressed put flag

Author: AlisinaDevelo

CHANGELOG.md

@@ -10,6 +10,7 @@ All notable changes to StreamHive are documented here. This project follows [Sem
 - **`storage`**: `BlobKeyLister` interface plus deterministic `ListKeys` support for memory and file stores.
 - **`replication`**: `blob.has`, `blob.get`, and `blob.missing` message types for anti-entropy sync.
 - **CLI**: startup anti-entropy sync for `-replicate` peers using `blob.has` / `blob.missing` / `blob.put`.
+- **CLI**: `-put-content-key` for sending blobs under `SHA-256(-put-data)` content keys.
 - **Demo**: 3-node Docker Compose demo with durable stores and node restart rehydration.
 - **Metrics**: Prometheus text endpoint at `/metrics/prometheus`.
 

README.md

@@ -35,7 +35,7 @@ go run . -listen 127.0.0.1:7070 -replicate -health 127.0.0.1:8080
 Terminal 2: dial the receiver and send one blob.
 
 ```bash
-go run . -listen 127.0.0.1:0 -dial 127.0.0.1:7070 -put-key demo -put-data "hello streamhive" -exit-after-put
+go run . -listen 127.0.0.1:0 -dial 127.0.0.1:7070 -put-content-key -put-data "hello streamhive" -exit-after-put
 ```
 
 Inspect counters:
@@ -44,7 +44,7 @@ Inspect counters:
 curl -s http://127.0.0.1:8080/metrics
 ```
 
-Look for `replication_blobs_stored`, `replication_bytes_stored`, and transport frame counters. Use `/metrics` for JSON or `/metrics/prometheus` for Prometheus text format.
+Look for `replication_blobs_stored`, `replication_bytes_stored`, and transport frame counters. The sender derives the blob key from `SHA-256(put-data)` when `-put-content-key` is set. Use `/metrics` for JSON or `/metrics/prometheus` for Prometheus text format.
 
 Or run the whole flow:
 
@@ -79,7 +79,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/replication` | Blob replication messages (`blob.put`) and store apply helper |
+| `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 |
 
 Wire handshake string constant: `p2p.HandshakeVersionV1` (carry inside application frames).
@@ -101,7 +101,8 @@ Wire handshake string constant: `p2p.HandshakeVersionV1` (carry inside applicati
 | `-tls-ca` / `-tls-server-name` / `-tls-insecure-skip-verify` | Client TLS |
 | `-replicate` | Enable blob replication from framed peers |
 | `-store-dir` | Persist replicated blobs with `storage.FileStore` |
-| `-put-key` / `-put-data` | Send one blob to the `-dial` peer |
+| `-put-key` / `-put-data` | Send one manually keyed blob to outbound peers |
+| `-put-content-key` | Derive the outbound blob key from `SHA-256(-put-data)` |
 | `-exit-after-put` | Exit after the outbound blob frame is written |
 | `-max-blob-bytes` | Cap replicated blob payload size |
 
@@ -116,7 +117,7 @@ flowchart TB
     T[TCPTransport]
     F[SHV1 frames]
     S[BlobStore]
-    R[replication blob.put]
+    R[replication blob.put / blob.has]
     CLI --> T
     T --> F
     F --> R

docker-compose.yml

@@ -73,8 +73,7 @@ services:
       - 127.0.0.1:0
       - -dial
       - node1:7070
-      - -put-key
-      - compose-demo
+      - -put-content-key
       - -put-data
       - hello from compose
       - -exit-after-put

docs/ARCHITECTURE.md

@@ -81,7 +81,7 @@ classDiagram
 - Listener and peer map share a mutex; the accept loop exits when the listener is closed.
 - `Close` stops new accepts, waits for the accept goroutine, then closes open peer connections. Peer goroutines remove themselves from the map on EOF / error via `unregisterPeer`.
 - Optional `FrameHandler` runs per frame on each peer session until error, context cancellation, or disconnect.
-- CLI replication installs a `FrameHandler` that decodes `blob.put` messages and writes to `MemoryStore` by default, or `FileStore` when `-store-dir` is set. Outbound `-put-key` / `-put-data` sends one frame after `-dial` or `-peers` connects.
+- CLI replication installs a `FrameHandler` that decodes `blob.put` messages and writes to `MemoryStore` by default, or `FileStore` when `-store-dir` is set. Outbound `-put-key` / `-put-data` sends one manually keyed frame after `-dial` or `-peers` connects; `-put-content-key` derives the key from `SHA-256(-put-data)`.
 - `-peer-reconnect` manages only static `-peers` targets. It retries failed dials with exponential backoff and schedules another retry when an outbound configured peer disconnects.
 - Replication peers advertise local keys on connect. Receivers reply with `blob.missing`, and owners send the requested blobs with `blob.put`.
 
@@ -122,7 +122,7 @@ sequenceDiagram
   participant Replication
   participant Store as BlobStore
 
-  Sender->>SenderTransport: -dial receiver + -put-key/-put-data
+  Sender->>SenderTransport: -dial receiver + -put-key/-put-data or -put-content-key
   SenderTransport->>ReceiverTransport: SHV1 frame carrying blob.put JSON
   ReceiverTransport->>Replication: FrameHandler(payload)
   Replication->>Replication: decode + validate limits
@@ -133,8 +133,7 @@ Implemented:
 
 - Static peer replication over `-dial` and comma-separated `-peers`.
 - Optional reconnect/backoff for `-peers`.
-- One message type: `blob.put`.
-- Sync message vocabulary: `blob.has`, `blob.get`, and `blob.missing`.
+- 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.

main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"flag"
@@ -56,7 +57,8 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) error {
 	storeDir := fs.String("store-dir", "", "directory for durable replicated blobs (requires -replicate)")
 	putKey := fs.String("put-key", "", "send one replicated blob key to -dial peer")
 	putData := fs.String("put-data", "", "send one replicated blob value to -dial peer")
-	exitAfterPut := fs.Bool("exit-after-put", false, "exit after sending -put-key to the dialed peer")
+	putContentKey := fs.Bool("put-content-key", false, "derive the replicated blob key from SHA-256(-put-data)")
+	exitAfterPut := fs.Bool("exit-after-put", false, "exit after sending one blob to outbound peers")
 	maxBlobBytes := fs.Int("max-blob-bytes", replication.DefaultMaxDataBytes, "max replicated blob payload bytes")
 
 	tlsCert := fs.String("tls-cert", "", "path to PEM certificate (enables TLS on listener)")
@@ -81,15 +83,19 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) error {
 		return err
 	}
 	peerTargets := combinePeerTargets(dialTarget, peerList)
-	if *putKey != "" && len(peerTargets) == 0 {
-		return fmt.Errorf("replication: -put-key requires -dial or -peers")
+	putRequested := *putKey != "" || *putContentKey
+	if putRequested && len(peerTargets) == 0 {
+		return fmt.Errorf("replication: -put-key or -put-content-key requires -dial or -peers")
+	}
+	if *putContentKey && *putKey != "" {
+		return fmt.Errorf("replication: -put-content-key cannot be combined with -put-key")
 	}
 	if *peerReconnect {
 		if len(peerList) == 0 {
 			return fmt.Errorf("peers: -peer-reconnect requires -peers")
 		}
-		if *putKey != "" {
-			return fmt.Errorf("replication: -peer-reconnect cannot be combined with -put-key")
+		if putRequested {
+			return fmt.Errorf("replication: -peer-reconnect cannot be combined with -put-key or -put-content-key")
 		}
 		if err := validateReconnectBackoff(*peerReconnectMin, *peerReconnectMax); err != nil {
 			return err
@@ -120,9 +126,12 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) error {
 		}
 	}
 	var putPayload []byte
+	var putKeyBytes []byte
+	var putKeyLabel string
 	var putResult chan error
-	if *putKey != "" {
-		putPayload, err = replication.EncodeBlobPut([]byte(*putKey), []byte(*putData), replLimits)
+	if putRequested {
+		putKeyBytes, putKeyLabel = resolvePutKey(*putKey, []byte(*putData), *putContentKey)
+		putPayload, err = replication.EncodeBlobPut(putKeyBytes, []byte(*putData), replLimits)
 		if err != nil {
 			return err
 		}
@@ -147,7 +156,7 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) error {
 			replMetrics.BlobsSent.Add(1)
 			replMetrics.BytesSent.Add(uint64(len(*putData)))
 			reportPutResult(putResult, nil)
-			log.Info("replicated blob sent", "remote", peer.RemoteAddr().String(), "key", *putKey, "bytes", len(*putData))
+			log.Info("replicated blob sent", "remote", peer.RemoteAddr().String(), "key", putKeyLabel, "bytes", len(*putData))
 		}
 		if keyLister != nil {
 			if err := sendBlobHas(ctx, peer, keyLister, replLimits, tr.MaxFrameBytes); err != nil {
@@ -311,7 +320,7 @@ func handleReplicationMessage(
 		}
 		metrics.BlobsStored.Add(1)
 		metrics.BytesStored.Add(uint64(len(msg.Data)))
-		attrs := []any{"remote", peer.RemoteAddr().String(), "key", string(msg.Key), "bytes", len(msg.Data)}
+		attrs := []any{"remote", peer.RemoteAddr().String(), "key", formatBlobKey(msg.Key), "bytes", len(msg.Data)}
 		if memoryStore != nil {
 			attrs = append(attrs, "blobs", memoryStore.Len())
 		}
@@ -391,11 +400,26 @@ func sendRequestedBlobs(
 		}
 		metrics.BlobsSent.Add(1)
 		metrics.BytesSent.Add(uint64(len(data)))
-		log.Info("replicated blob sent", "remote", peer.RemoteAddr().String(), "key", string(key), "bytes", len(data))
+		log.Info("replicated blob sent", "remote", peer.RemoteAddr().String(), "key", formatBlobKey(key), "bytes", len(data))
 	}
 	return nil
 }
 
+func resolvePutKey(key string, data []byte, contentKey bool) ([]byte, string) {
+	if contentKey {
+		digest := storage.SHA256Key(data)
+		return digest, storage.SHA256KeyHex(data)
+	}
+	return []byte(key), key
+}
+
+func formatBlobKey(key []byte) string {
+	if err := storage.ValidateSHA256Key(key); err == nil {
+		return hex.EncodeToString(key)
+	}
+	return string(key)
+}
+
 func writePeerFrame(peer p2p.Peer, payload []byte, maxFrameBytes int) error {
 	tcpPeer, ok := peer.(*p2p.TCPPeer)
 	if !ok {

main_test.go

@@ -123,7 +123,19 @@ func TestRun_putRequiresDial(t *testing.T) {
 	var out bytes.Buffer
 	err := run(context.Background(), []string{"-put-key", "alpha", "-put-data", "hello"}, &out, io.Discard)
 	require.Error(t, err)
-	assert.Contains(t, err.Error(), "-put-key requires -dial or -peers")
+	assert.Contains(t, err.Error(), "-put-key or -put-content-key requires -dial or -peers")
+}
+
+func TestRun_putContentKeyRejectsExplicitKey(t *testing.T) {
+	var out bytes.Buffer
+	err := run(context.Background(), []string{
+		"-dial", "127.0.0.1:1",
+		"-put-key", "alpha",
+		"-put-content-key",
+		"-put-data", "hello",
+	}, &out, io.Discard)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "-put-content-key cannot be combined with -put-key")
 }
 
 func TestRun_storeDirRequiresReplicate(t *testing.T) {
@@ -240,6 +252,54 @@ func TestRun_replicatesBlobPutToFileStore(t *testing.T) {
 	require.NoError(t, <-serverErrCh)
 }
 
+func TestRun_replicatesContentKeyedBlobPutToFileStore(t *testing.T) {
+	storeDir := t.TempDir()
+	serverCtx, serverCancel := context.WithCancel(context.Background())
+	defer serverCancel()
+	var serverOut, serverErr safeBuffer
+	serverErrCh := make(chan error, 1)
+	go func() {
+		serverErrCh <- run(serverCtx, []string{
+			"-listen", "127.0.0.1:0",
+			"-replicate",
+			"-store-dir", storeDir,
+		}, &serverOut, &serverErr)
+	}()
+
+	require.Eventually(t, func() bool {
+		return strings.Contains(serverOut.String(), "listening on")
+	}, 3*time.Second, 20*time.Millisecond)
+
+	re := regexp.MustCompile(`listening on ([^\n]+)`)
+	m := re.FindStringSubmatch(serverOut.String())
+	require.Len(t, m, 2, "stdout=%q", serverOut.String())
+
+	data := []byte("address me by content")
+	var clientOut, clientErr safeBuffer
+	err := run(context.Background(), []string{
+		"-listen", "127.0.0.1:0",
+		"-dial", m[1],
+		"-put-content-key",
+		"-put-data", string(data),
+		"-exit-after-put",
+	}, &clientOut, &clientErr)
+	require.NoError(t, err, "client logs=%q", clientErr.String())
+
+	key := storage.SHA256Key(data)
+	require.Eventually(t, func() bool {
+		store, err := storage.NewFileStore(storeDir)
+		if err != nil {
+			return false
+		}
+		got, err := store.Get(context.Background(), key)
+		return err == nil && string(got) == string(data)
+	}, 3*time.Second, 20*time.Millisecond, "server logs=%q", serverErr.String())
+	assert.Contains(t, serverErr.String(), storage.SHA256KeyHex(data))
+
+	serverCancel()
+	require.NoError(t, <-serverErrCh)
+}
+
 func TestRun_syncsMissingBlobOnConnect(t *testing.T) {
 	sourceDir := t.TempDir()
 	targetDir := t.TempDir()
@@ -358,6 +418,22 @@ func TestMissingKeys(t *testing.T) {
 	require.Equal(t, [][]byte{[]byte("a")}, again)
 }
 
+func TestResolvePutKey(t *testing.T) {
+	key, label := resolvePutKey("manual", []byte("hello"), false)
+	assert.Equal(t, []byte("manual"), key)
+	assert.Equal(t, "manual", label)
+
+	key, label = resolvePutKey("", []byte("hello"), true)
+	assert.Equal(t, storage.SHA256Key([]byte("hello")), key)
+	assert.Equal(t, storage.SHA256KeyHex([]byte("hello")), label)
+}
+
+func TestFormatBlobKey(t *testing.T) {
+	data := []byte("hello")
+	assert.Equal(t, "manual", formatBlobKey([]byte("manual")))
+	assert.Equal(t, storage.SHA256KeyHex(data), formatBlobKey(storage.SHA256Key(data)))
+}
+
 func TestWritePrometheusMetrics(t *testing.T) {
 	var out bytes.Buffer
 	writePrometheusMetrics(&out, map[string]int64{

scripts/demo-replication.sh

@@ -5,7 +5,6 @@ ROOT_DIR="$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)"
 BIN="$ROOT_DIR/bin/fs"
 P2P_ADDR="${P2P_ADDR:-127.0.0.1:7070}"
 HEALTH_ADDR="${HEALTH_ADDR:-127.0.0.1:8080}"
-BLOB_KEY="${BLOB_KEY:-demo}"
 BLOB_DATA="${BLOB_DATA:-hello streamhive}"
 LOG_FILE="$(mktemp -t streamhive-demo.XXXXXX)"
 
@@ -39,7 +38,7 @@ done
 "$BIN" \
 	-listen 127.0.0.1:0 \
 	-dial "$P2P_ADDR" \
-	-put-key "$BLOB_KEY" \
+	-put-content-key \
 	-put-data "$BLOB_DATA" \
 	-exit-after-put
 
@@ -54,6 +53,6 @@ until curl -fsS "http://$HEALTH_ADDR/metrics" | grep '"replication_blobs_stored"
 	sleep 0.1
 done
 
-echo "replicated blob '$BLOB_KEY' (${#BLOB_DATA} bytes)"
+echo "replicated content-addressed blob (${#BLOB_DATA} bytes)"
 curl -fsS "http://$HEALTH_ADDR/metrics"
 echo