fix: address code review findings for shared hub

1. Fix listen command: now streams via Listen RPC instead of
   blocking after initial sync
2. Add input validation on Publish: type, ID, origin, content
   size limit (1MB)
3. Warn on --share publish failure instead of silent suppression
4. Constant-time token comparison via crypto/subtle + O(1) map
   lookup
5. Wire Raft cluster to Server with SetCluster/Shutdown
6. Reject duplicate project registration in store
7. Disconnect slow fanout listeners instead of silently dropping
8. File locking on sync state to prevent concurrent race
9. Fail fast on auth errors in failover client

Signed-off-by: Murat Parlakisik <parlakisik@gmail.com>

Author: parlakisik

internal/cli/add/cmd/root/run.go

@@ -25,6 +25,7 @@ import (
 	"github.com/ActiveMemory/ctx/internal/hub"
 	"github.com/ActiveMemory/ctx/internal/trace"
 	writeAdd "github.com/ActiveMemory/ctx/internal/write/add"
+	writeConnect "github.com/ActiveMemory/ctx/internal/write/connect"
 )
 
 // Run executes the add command logic.
@@ -87,9 +88,11 @@ func Run(cmd *cobra.Command, args []string, flags entity.AddConfig) error {
 			Content: content,
 			Origin:  filepath.Base(state.Dir()),
 		}
-		_ = corePub.Run(
+		if pubErr := corePub.Run(
 			cmd, []hub.PublishEntry{pubEntry},
-		)
+		); pubErr != nil {
+			writeConnect.PublishFailed(cmd, pubErr)
+		}
 	}
 
 	if fType == cfgEntry.Task && coreEntry.NeedsSpec(content) {

internal/cli/connect/core/listen/listen.go

@@ -19,14 +19,13 @@ import (
 	writeConnect "github.com/ActiveMemory/ctx/internal/write/connect"
 )
 
-// Run streams entries from the hub in real-time.
-//
-// Connects to the hub, starts a Sync to get recent entries,
-// then blocks waiting for new entries. Writes each entry to
-// .context/shared/ as it arrives. Stops on Ctrl-C.
+// Run streams entries from the hub in real-time via the
+// Listen RPC. Writes each entry to .context/shared/ as
+// it arrives. Stops on Ctrl-C.
 //
 // Parameters:
 //   - cmd: cobra command for output
+//   - args: unused (cobra signature)
 //
 // Returns:
 //   - error: non-nil if config, connection, or write fails
@@ -51,21 +50,23 @@ func Run(cmd *cobra.Command, _ []string) error {
 
 	writeConnect.Listening(cmd)
 
-	entries, syncErr := client.Sync(
+	listenErr := client.Listen(
 		ctx, cfg.Types, 0,
+		func(msg hub.EntryMsg) error {
+			writeErr := render.WriteEntries(
+				[]hub.EntryMsg{msg},
+			)
+			if writeErr != nil {
+				return writeErr
+			}
+			writeConnect.EntryReceived(cmd, msg.Type)
+			return nil
+		},
 	)
-	if syncErr != nil {
-		return syncErr
-	}
-	if len(entries) > 0 {
-		if writeErr := render.WriteEntries(
-			entries,
-		); writeErr != nil {
-			return writeErr
-		}
-		writeConnect.Synced(cmd, len(entries))
-	}
 
-	<-ctx.Done()
+	// Context cancellation (Ctrl-C) is expected.
+	if listenErr != nil && ctx.Err() == nil {
+		return listenErr
+	}
 	return nil
 }

internal/cli/connect/core/sync/state.go

@@ -19,33 +19,58 @@ import (
 // stateFile is the sync state persistence filename.
 const stateFile = ".sync-state.json"
 
+// lockFile is the lock file to prevent concurrent syncs.
+const lockFile = ".sync.lock"
+
 // loadState reads sync state from .context/shared/.
-func loadState() (state, error) {
+// Acquires a lock file to prevent concurrent access.
+func loadState() (state, func(), error) {
 	var s state
-	path := filepath.Join(
-		rc.ContextDir(), "shared", stateFile,
-	)
+	dir := filepath.Join(rc.ContextDir(), "shared")
+	lockPath := filepath.Join(dir, lockFile)
+
+	if mkErr := io.SafeMkdirAll(
+		dir, fs.PermKeyDir,
+	); mkErr != nil {
+		return s, nil, mkErr
+	}
+
+	// Acquire lock: fail if another sync is running.
+	if _, statErr := os.Stat(lockPath); statErr == nil {
+		return s, nil, os.ErrExist
+	}
+	if writeErr := io.SafeWriteFile(
+		lockPath, []byte("lock"), fs.PermFile,
+	); writeErr != nil {
+		return s, nil, writeErr
+	}
+
+	release := func() { _ = os.Remove(lockPath) }
+
+	path := filepath.Join(dir, stateFile)
 	data, readErr := io.SafeReadUserFile(path)
 	if os.IsNotExist(readErr) {
-		return s, nil
+		return s, release, nil
 	}
 	if readErr != nil {
-		return s, readErr
+		release()
+		return s, nil, readErr
 	}
 	if len(data) == 0 {
-		return s, nil
+		return s, release, nil
+	}
+	if unmarshalErr := json.Unmarshal(
+		data, &s,
+	); unmarshalErr != nil {
+		release()
+		return s, nil, unmarshalErr
 	}
-	return s, json.Unmarshal(data, &s)
+	return s, release, nil
 }
 
 // saveState writes sync state to .context/shared/.
 func saveState(s state) error {
 	dir := filepath.Join(rc.ContextDir(), "shared")
-	if mkErr := io.SafeMkdirAll(
-		dir, fs.PermKeyDir,
-	); mkErr != nil {
-		return mkErr
-	}
 	data, marshalErr := json.MarshalIndent(
 		s, "", "  ",
 	)

internal/cli/connect/core/sync/sync.go

@@ -33,10 +33,11 @@ func Run(cmd *cobra.Command) error {
 		return loadErr
 	}
 
-	syncState, stateErr := loadState()
+	syncState, releaseLock, stateErr := loadState()
 	if stateErr != nil {
 		return stateErr
 	}
+	defer releaseLock()
 
 	client, dialErr := hub.NewClient(
 		cfg.HubAddr, cfg.Token,

internal/cli/serve/core/shared/shared.go

@@ -76,20 +76,21 @@ func Run(
 		return tokenErr
 	}
 
+	srv := hub.NewServer(store, adminToken)
+
 	// Start Raft cluster if peers are configured.
 	if len(peers) > 0 {
 		bindAddr := fmt.Sprintf(":%d", port+1)
-		_, clusterErr := hub.NewCluster(
+		cluster, clusterErr := hub.NewCluster(
 			fmt.Sprintf(":%d", port),
 			bindAddr, dataDir, peers,
 		)
 		if clusterErr != nil {
 			return clusterErr
 		}
+		srv.SetCluster(cluster)
 	}
 
-	srv := hub.NewServer(store, adminToken)
-
 	addr := fmt.Sprintf(":%d", port)
 	lis, lisErr := net.Listen("tcp", addr)
 	if lisErr != nil {

internal/err/hub/hub.go

@@ -39,6 +39,29 @@ func InternalErr(cause error) error {
 //
 // Returns:
 //   - error: "action must be 'add' or 'remove', got <action>"
+//
+// DuplicateProject returns an error when a project is
+// already registered.
+//
+// Parameters:
+//   - name: the duplicate project name
+//
+// Returns:
+//   - error: "project already registered: <name>"
+func DuplicateProject(name string) error {
+	return fmt.Errorf(
+		"project already registered: %q", name,
+	)
+}
+
+// InvalidPeerAction returns an error for an unrecognized
+// peer action.
+//
+// Parameters:
+//   - action: the unrecognized action string
+//
+// Returns:
+//   - error: formatted error with the invalid action
 func InvalidPeerAction(action string) error {
 	return fmt.Errorf(
 		"action must be 'add' or 'remove', got %q",

internal/hub/client.go

@@ -139,6 +139,57 @@ func (c *Client) Sync(
 	return entries, nil
 }
 
+// Listen opens a server-streaming Listen RPC and calls
+// the handler for each entry received. Blocks until the
+// context is cancelled or the stream ends.
+//
+// Parameters:
+//   - ctx: context for cancellation
+//   - types: entry types to receive (empty = all)
+//   - sinceSequence: start from this sequence
+//   - handler: called for each received entry
+//
+// Returns:
+//   - error: non-nil if stream setup or recv fails
+func (c *Client) Listen(
+	ctx context.Context,
+	types []string,
+	sinceSequence uint64,
+	handler func(EntryMsg) error,
+) error {
+	stream, streamErr := c.conn.NewStream(
+		c.authedCtx(ctx),
+		&grpc.StreamDesc{ServerStreams: true},
+		"/ctx.hub.v1.CtxHub/Listen",
+	)
+	if streamErr != nil {
+		return streamErr
+	}
+
+	if sendErr := stream.SendMsg(&ListenRequest{
+		Types:         types,
+		SinceSequence: sinceSequence,
+	}); sendErr != nil {
+		return sendErr
+	}
+	if closeErr := stream.CloseSend(); closeErr != nil {
+		return closeErr
+	}
+
+	for {
+		msg := &EntryMsg{}
+		if recvErr := stream.RecvMsg(msg); recvErr != nil {
+			if isEOF(recvErr) {
+				return nil
+			}
+			return recvErr
+		}
+		if handleErr := handler(*msg); handleErr != nil {
+			return handleErr
+		}
+	}
+}
+
 // Status calls the Status RPC.
 //
 // Parameters:

internal/hub/entry_validate.go

@@ -0,0 +1,52 @@
+//   /    ctx:                         https://ctx.ist
+// ,'`./    do you remember?
+// `.,'\
+//   \    Copyright 2026-present Context contributors.
+//                 SPDX-License-Identifier: Apache-2.0
+
+package hub
+
+import (
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// maxContentLen is the maximum entry content size (1MB).
+const maxContentLen = 1 << 20
+
+// allowedTypes is the set of valid entry types.
+var allowedTypes = map[string]bool{
+	"decision":   true,
+	"learning":   true,
+	"convention": true,
+	"task":       true,
+}
+
+// validateEntry checks a PublishEntry for required fields
+// and enforces size limits.
+func validateEntry(pe PublishEntry) error {
+	if pe.ID == "" {
+		return status.Error(
+			codes.InvalidArgument, "entry ID required",
+		)
+	}
+	if !allowedTypes[pe.Type] {
+		return status.Errorf(
+			codes.InvalidArgument,
+			"invalid entry type %q", pe.Type,
+		)
+	}
+	if pe.Origin == "" {
+		return status.Error(
+			codes.InvalidArgument,
+			"entry origin required",
+		)
+	}
+	if len(pe.Content) > maxContentLen {
+		return status.Error(
+			codes.InvalidArgument,
+			"entry content exceeds 1MB limit",
+		)
+	}
+	return nil
+}

internal/hub/errcheck.go

@@ -0,0 +1,24 @@
+//   /    ctx:                         https://ctx.ist
+// ,'`./    do you remember?
+// `.,'\
+//   \    Copyright 2026-present Context contributors.
+//                 SPDX-License-Identifier: Apache-2.0
+
+package hub
+
+import (
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// isAuthErr reports whether err is an authentication or
+// authorization failure.
+func isAuthErr(err error) bool {
+	s, ok := status.FromError(err)
+	if !ok {
+		return false
+	}
+	c := s.Code()
+	return c == codes.Unauthenticated ||
+		c == codes.PermissionDenied
+}

internal/hub/failover.go

@@ -14,8 +14,8 @@ import (
 )
 
 // NewFailoverClient creates a client that tries peers in
-// order until one succeeds. The first address is the
-// primary; others are fallbacks.
+// order until one succeeds. Fails fast on auth errors
+// since the token is the same for all peers.
 //
 // Parameters:
 //   - peers: ordered list of hub addresses
@@ -43,7 +43,6 @@ func NewFailoverClient(
 			continue
 		}
 
-		// Verify connectivity with a Status call.
 		resp := &StatusResponse{}
 		callErr := conn.Invoke(
 			addBearerMD(
@@ -55,6 +54,13 @@ func NewFailoverClient(
 		)
 		if callErr != nil {
 			_ = conn.Close()
+
+			// Fail fast on auth errors — same token
+			// won't work on other peers either.
+			if isAuthErr(callErr) {
+				return nil, callErr
+			}
+
 			lastErr = callErr
 			continue
 		}