From ed3d4d253dcca8aceb3647c49f6dceb667f9a377 Mon Sep 17 00:00:00 2001 From: Omer Kocaoglu Date: Tue, 7 Jul 2026 08:22:19 -0400 Subject: [PATCH] fix(hub): surface silently swallowed errors in hubsync hook and replication loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two hub code paths swallowed errors with no logging surface — the session-start hubsync hook and the cluster replication loop. Operators could not tell a healthy quiet sync from one that never reached the network. Route every silent return through the warn sink with format constants in config/warn, and un-conflate the sync-error check from the empty-result check so only a real error warns; a genuine empty result stays silent. Bound the session-start pull with a 10s deadline (HubSyncTimeout) so a hub that accepts the connection but never responds cannot stall session start; the hook still returns "" on timeout like any other failure. The replication recv loop stays silent on io.EOF (normal stream end) and caller shutdown, and warns on anything else. Both functions keep their signatures and never-block contracts; the logging and the deadline are the only behavior changes. Resubmission of the work from PR #114, which was swept closed during the DCO history-rewrite of main rather than rejected. Incorporates the final review round: - Drop the memory-file hunk from the PR; the deferred replication warn-cadence follow-up is recorded on main post-merge. - Fix the "next session retries" comments — the throttle marker is stamped unconditionally after the pull, so a cut-off pull retries on the next day's first session, not the next session. - Correct the Sync doc comment: it returns only a status string. - Keep the eof helper strict (==) for the clean-end callers in client.go; broaden to errors.Is inline only at the replication warn-suppression site, so each polarity gets the matching strictness. - Register the warn capture before the black-hole listener in the hung-hub test so LIFO cleanup cannot leak a late warning from a stranded goroutine into the next test. Closes #100. Spec: specs/fix-hub-silent-error-suppression.md Signed-off-by: Omer Kocaoglu --- internal/cli/system/core/hubsync/doc.go | 9 +- .../cli/system/core/hubsync/export_test.go | 19 ++ internal/cli/system/core/hubsync/sync.go | 39 ++- internal/cli/system/core/hubsync/sync_test.go | 260 ++++++++++++++++++ internal/config/hub/doc.go | 2 + internal/config/hub/hub.go | 15 + internal/config/warn/warn.go | 51 ++++ internal/hub/eof.go | 18 +- internal/hub/replicate.go | 20 ++ internal/hub/replicate_test.go | 203 ++++++++++++++ specs/fix-hub-silent-error-suppression.md | 160 +++++++++++ 11 files changed, 785 insertions(+), 11 deletions(-) create mode 100644 internal/cli/system/core/hubsync/export_test.go create mode 100644 internal/cli/system/core/hubsync/sync_test.go create mode 100644 internal/hub/replicate_test.go create mode 100644 specs/fix-hub-silent-error-suppression.md diff --git a/internal/cli/system/core/hubsync/doc.go b/internal/cli/system/core/hubsync/doc.go index 07f821077..6303396f4 100644 --- a/internal/cli/system/core/hubsync/doc.go +++ b/internal/cli/system/core/hubsync/doc.go @@ -23,8 +23,13 @@ // writes them to disk via the connection render layer. // It returns a formatted status message with the count // of synced entries, or an empty string when nothing -// was fetched. Every error is silently swallowed so the -// hook never blocks the session start. +// was fetched. Every error is surfaced as a stderr +// warning via the warn sink, but never propagates: the +// hook must not block the session start. The pull is +// bounded by [github.com/ActiveMemory/ctx/internal/config/hub.HubSyncTimeout] +// so a hub that accepts the connection but never responds +// degrades to a warning rather than stalling the hook. +// Only a genuine zero-entry result stays silent. // // The data flow is: // diff --git a/internal/cli/system/core/hubsync/export_test.go b/internal/cli/system/core/hubsync/export_test.go new file mode 100644 index 000000000..12894524e --- /dev/null +++ b/internal/cli/system/core/hubsync/export_test.go @@ -0,0 +1,19 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package hubsync + +import "time" + +// SetSyncTimeoutForTest overrides the session-start pull deadline +// and returns a restore func. It lets external tests drive the +// deadline against an unresponsive hub in milliseconds instead of +// the full production interval. +func SetSyncTimeoutForTest(d time.Duration) func() { + prev := syncTimeout + syncTimeout = d + return func() { syncTimeout = prev } +} diff --git a/internal/cli/system/core/hubsync/sync.go b/internal/cli/system/core/hubsync/sync.go index 1d918377a..bfdbb211f 100644 --- a/internal/cli/system/core/hubsync/sync.go +++ b/internal/cli/system/core/hubsync/sync.go @@ -12,6 +12,7 @@ import ( "fmt" "os" "path/filepath" + "time" "github.com/ActiveMemory/ctx/internal/assets/read/desc" connectCfg "github.com/ActiveMemory/ctx/internal/cli/connection/core/config" @@ -53,10 +54,17 @@ func Connected(ctxDir string) (bool, error) { return true, nil } +// syncTimeout bounds the session-start pull RPC. It is a var, +// not the bare constant, so tests can shrink it to exercise the +// deadline against a deliberately unresponsive hub without +// waiting the full production interval. +var syncTimeout = time.Duration(cfgHub.HubSyncTimeout) * time.Second + // Sync pulls new entries from the hub and writes them to -// .context/hub/. Returns the count of synced entries -// and a formatted status message, or empty string if no -// new entries. +// .context/hub/. Returns a formatted status message, or +// empty string if nothing was synced (no new entries, or +// any failure — every failure is surfaced as a warning and +// still returns "" so the hook never blocks). // // Parameters: // - sessionID: current session ID (unused, for future) @@ -66,6 +74,7 @@ func Connected(ctxDir string) (bool, error) { func Sync(_ string) string { cfg, loadErr := connectCfg.Load() if loadErr != nil { + logWarn.Warn(cfgWarn.HubSyncLoadConfig, loadErr) return "" } @@ -73,6 +82,7 @@ func Sync(_ string) string { cfg.HubAddr, cfg.Token, ) if dialErr != nil { + logWarn.Warn(cfgWarn.HubSyncDial, cfg.HubAddr, dialErr) return "" } defer func() { @@ -81,14 +91,31 @@ func Sync(_ string) string { } }() - entries, syncErr := client.Sync( - context.Background(), cfg.Types, 0, + // Bound the pull with a deadline so a hub that accepts the + // connection but never responds (hung server, black-hole + // proxy) cannot stall session start: the RPC has no inherent + // timeout, and this hook must never block. An exceeded + // deadline surfaces as a warning like any other sync failure. + // The throttle marker is stamped unconditionally after the + // pull, so a cut-off pull retries on the next day's first + // session, not the next session. + ctx, cancel := context.WithTimeout( + context.Background(), syncTimeout, ) - if syncErr != nil || len(entries) == 0 { + defer cancel() + + entries, syncErr := client.Sync(ctx, cfg.Types, 0) + if syncErr != nil { + logWarn.Warn(cfgWarn.HubSyncPull, cfg.HubAddr, syncErr) + return "" + } + if len(entries) == 0 { + // Genuine empty result: not an error, no warning. return "" } if writeErr := render.WriteEntries(entries); writeErr != nil { + logWarn.Warn(cfgWarn.HubSyncWrite, len(entries), writeErr) return "" } diff --git a/internal/cli/system/core/hubsync/sync_test.go b/internal/cli/system/core/hubsync/sync_test.go new file mode 100644 index 000000000..b6cffe8d6 --- /dev/null +++ b/internal/cli/system/core/hubsync/sync_test.go @@ -0,0 +1,260 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package hubsync_test + +import ( + "bytes" + "net" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/ActiveMemory/ctx/internal/assets/read/lookup" + connectCfg "github.com/ActiveMemory/ctx/internal/cli/connection/core/config" + "github.com/ActiveMemory/ctx/internal/cli/system/core/hubsync" + "github.com/ActiveMemory/ctx/internal/config/fs" + "github.com/ActiveMemory/ctx/internal/crypto" + "github.com/ActiveMemory/ctx/internal/hub" + logWarn "github.com/ActiveMemory/ctx/internal/log/warn" + "github.com/ActiveMemory/ctx/internal/testutil/testctx" +) + +// TestMain initializes the embedded text-asset lookup so error +// strings rendered into warnings resolve their DescKey text. +func TestMain(m *testing.M) { + lookup.Init() + os.Exit(m.Run()) +} + +// declareContext positions the test in a temp project with a +// materialized .context/ directory and returns its path. +func declareContext(t *testing.T) string { + t.Helper() + dir := t.TempDir() + ctxDir := testctx.Declare(t, dir) + if mkErr := os.MkdirAll(ctxDir, fs.PermExec); mkErr != nil { + t.Fatal(mkErr) + } + return ctxDir +} + +// captureWarnings redirects the warn sink to a buffer for the +// duration of the test. +func captureWarnings(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + restore := logWarn.SetSink(&buf) + t.Cleanup(restore) + return &buf +} + +// saveConnectConfig generates a global encryption key under the +// test HOME and persists a connection config pointing at addr. +func saveConnectConfig(t *testing.T, addr, token string) { + t.Helper() + key, keyErr := crypto.GenerateKey() + if keyErr != nil { + t.Fatal(keyErr) + } + keyPath := crypto.GlobalKeyPath() + if mkErr := os.MkdirAll( + filepath.Dir(keyPath), fs.PermKeyDir, + ); mkErr != nil { + t.Fatal(mkErr) + } + if saveErr := crypto.SaveKey(keyPath, key); saveErr != nil { + t.Fatal(saveErr) + } + if cfgErr := connectCfg.Save(connectCfg.Config{ + HubAddr: addr, + Token: token, + }); cfgErr != nil { + t.Fatal(cfgErr) + } +} + +// startHub serves a hub with the given store on a random port +// and returns its address and a registered client token. +func startHub(t *testing.T, store *hub.Store) (string, string) { + t.Helper() + adminTok, tokErr := hub.GenerateAdminToken() + if tokErr != nil { + t.Fatal(tokErr) + } + srv := hub.NewServer(store, adminTok) + lis, lisErr := net.Listen("tcp", "127.0.0.1:0") + if lisErr != nil { + t.Fatal(lisErr) + } + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.GracefulStop) + + addr := lis.Addr().String() + client, dialErr := hub.NewClient(addr, "") + if dialErr != nil { + t.Fatal(dialErr) + } + defer func() { + if cerr := client.Close(); cerr != nil { + t.Log(cerr) + } + }() + reg, regErr := client.Register( + t.Context(), adminTok, "hubsync-test", + ) + if regErr != nil { + t.Fatal(regErr) + } + return addr, reg.ClientToken +} + +func TestSync_WarnsOnLoadError(t *testing.T) { + declareContext(t) + buf := captureWarnings(t) + + if got := hubsync.Sync(""); got != "" { + t.Errorf("Sync = %q, want empty on load failure", got) + } + if !strings.Contains( + buf.String(), "hubsync: load connection config:", + ) { + t.Errorf("missing load warning, got: %q", buf.String()) + } +} + +func TestSync_WarnsOnDialError(t *testing.T) { + declareContext(t) + // grpc.NewClient is lazy for almost every bad target, but + // a control character fails URL parsing at construction — + // the one eager failure mode, and exactly what a corrupted + // connect config would produce. + saveConnectConfig(t, "\x00", "tok") + buf := captureWarnings(t) + + if got := hubsync.Sync(""); got != "" { + t.Errorf("Sync = %q, want empty on dial failure", got) + } + if !strings.Contains(buf.String(), "hubsync: dial") { + t.Errorf("missing dial warning, got: %q", buf.String()) + } +} + +func TestSync_WarnsOnPullError(t *testing.T) { + declareContext(t) + // A well-formed address nobody listens on: client + // construction is lazy, so the failure surfaces at the + // Sync RPC. + lis, lisErr := net.Listen("tcp", "127.0.0.1:0") + if lisErr != nil { + t.Fatal(lisErr) + } + addr := lis.Addr().String() + if closeErr := lis.Close(); closeErr != nil { + t.Fatal(closeErr) + } + saveConnectConfig(t, addr, "tok") + buf := captureWarnings(t) + + if got := hubsync.Sync(""); got != "" { + t.Errorf("Sync = %q, want empty on pull failure", got) + } + if !strings.Contains(buf.String(), "hubsync: sync from") { + t.Errorf("missing pull warning, got: %q", buf.String()) + } +} + +func TestSync_NoWarnOnEmptyResult(t *testing.T) { + ctxDir := declareContext(t) + store, storeErr := hub.NewStore( + filepath.Join(ctxDir, "hub-data"), + ) + if storeErr != nil { + t.Fatal(storeErr) + } + addr, token := startHub(t, store) + saveConnectConfig(t, addr, token) + buf := captureWarnings(t) + + if got := hubsync.Sync(""); got != "" { + t.Errorf("Sync = %q, want empty for zero entries", got) + } + if buf.Len() != 0 { + t.Errorf( + "empty result must not warn, got: %q", buf.String(), + ) + } +} + +// blackHoleListener accepts TCP connections but never speaks +// gRPC, so a dial succeeds while every RPC blocks until its +// deadline. It models a hung-but-reachable hub. Accepted +// connections and the listener are closed at test cleanup. +func blackHoleListener(t *testing.T) string { + t.Helper() + lis, lisErr := net.Listen("tcp", "127.0.0.1:0") + if lisErr != nil { + t.Fatal(lisErr) + } + var mu sync.Mutex + var conns []net.Conn + t.Cleanup(func() { + _ = lis.Close() + mu.Lock() + for _, c := range conns { + _ = c.Close() + } + mu.Unlock() + }) + go func() { + for { + conn, acceptErr := lis.Accept() + if acceptErr != nil { + return + } + mu.Lock() + conns = append(conns, conn) + mu.Unlock() + } + }() + return lis.Addr().String() +} + +func TestSync_WarnsOnHungHub(t *testing.T) { + declareContext(t) + // Capture warnings before standing up the black-hole listener. + // Cleanup is LIFO, so registering the warn-sink restore first + // makes it run last — after the listener's connections close. On + // the failure path (deadline not enforced), that ordering lets a + // leaked Sync goroutine's late warning land in this test's buffer + // rather than the restored process-global sink of the next test. + buf := captureWarnings(t) + // A hub that accepts the connection but never responds. The + // RPC has no inherent deadline, so without HubSyncTimeout this + // would hang the session-start hook indefinitely. Shrink the + // bound so the test exercises the deadline in milliseconds. + saveConnectConfig(t, blackHoleListener(t), "tok") + restore := hubsync.SetSyncTimeoutForTest(200 * time.Millisecond) + t.Cleanup(restore) + + done := make(chan string, 1) + go func() { done <- hubsync.Sync("") }() + + select { + case got := <-done: + if got != "" { + t.Errorf("Sync = %q, want empty on deadline", got) + } + case <-time.After(5 * time.Second): + t.Fatal("Sync did not return: the deadline was not enforced") + } + if !strings.Contains(buf.String(), "hubsync: sync from") { + t.Errorf("missing pull warning, got: %q", buf.String()) + } +} diff --git a/internal/config/hub/doc.go b/internal/config/hub/doc.go index 0eb87a683..63dd522b5 100644 --- a/internal/config/hub/doc.go +++ b/internal/config/hub/doc.go @@ -91,6 +91,8 @@ // - RoleFollower, RoleActive: status role labels // - ReplicateInterval (5s): follower retry // interval +// - HubSyncTimeout (10s): session-start pull +// deadline // - ThrottleHubSync: daily sync throttle marker // // # Why Centralized diff --git a/internal/config/hub/hub.go b/internal/config/hub/hub.go index 0d8fe4e70..94dc5ffe2 100644 --- a/internal/config/hub/hub.go +++ b/internal/config/hub/hub.go @@ -76,6 +76,21 @@ const ( ReplicateInterval = 5 // seconds ) +// Session-start sync timing. +const ( + // HubSyncTimeout bounds the session-start hubsync pull RPC. + // The hook must never block session start, but the Sync RPC + // has no inherent deadline, so a hub that accepts the + // connection yet never responds would hang the hook + // indefinitely. The bound is generous enough for a healthy + // pull over a slow link and short enough that a hung hub + // degrades to a warning instead of a stall; the daily + // throttle plus next-day retry covers a cut-off pull (the + // throttle marker is stamped unconditionally, so retry is + // bounded to the next day's first session, not the next). + HubSyncTimeout = 10 // seconds +) + // Token generation. const ( // TokenBytes is the number of random bytes in a diff --git a/internal/config/warn/warn.go b/internal/config/warn/warn.go index daa9d9999..f183aaf8f 100644 --- a/internal/config/warn/warn.go +++ b/internal/config/warn/warn.go @@ -120,6 +120,33 @@ const ( // the loss visible. HubReplicateAppend = "hub replicate append: %v" + // HubReplicateDial is the stderr format for a failed gRPC + // client construction toward the master. Takes (masterAddr, + // error). Like every replication warning, it fires once per + // attempt; the loop retries on its own interval. + HubReplicateDial = "hub replicate dial %s: %v" + + // HubReplicateStream is the stderr format for a failed sync + // stream open toward the master. Takes (masterAddr, error). + HubReplicateStream = "hub replicate open stream %s: %v" + + // HubReplicateSend is the stderr format for a failed sync + // request send on the replication stream. Takes (masterAddr, + // error). + HubReplicateSend = "hub replicate send request %s: %v" + + // HubReplicateCloseSend is the stderr format for a failed + // half-close of the replication stream. Takes (masterAddr, + // error). + HubReplicateCloseSend = "hub replicate close send %s: %v" + + // HubReplicateRecv is the stderr format for a transport + // failure while receiving replicated entries. Takes + // (masterAddr, error). io.EOF (the normal end of a sync + // stream) and caller shutdown are deliberately not warned; + // see replicateOnce. + HubReplicateRecv = "hub replicate recv %s: %v" + // StateInitializedProbe is the stderr format for failures // inside [state.Initialized] beyond "no context dir declared." // Hooks bail on false either way, but a visible warning shows @@ -187,6 +214,30 @@ const ( NotifyWebhookPost = "notify: webhook POST failed: %v" ) +// Hubsync hook warning formats. The session-start hubsync hook +// must never block or fail the session, so [hubsync.Sync] keeps +// returning a (possibly empty) nudge string — these warnings are +// the only signal that a configured hub sync went wrong instead +// of merely finding nothing new. +const ( + // HubSyncLoadConfig is the format for a failed connection + // config load. Takes (error). + HubSyncLoadConfig = "hubsync: load connection config: %v" + + // HubSyncDial is the format for a rejected hub address at + // client construction. Takes (addr, error). + HubSyncDial = "hubsync: dial %s: %v" + + // HubSyncPull is the format for a failed Sync RPC. Takes + // (addr, error). A genuine zero-entry result is not an + // error and is never warned. + HubSyncPull = "hubsync: sync from %s: %v" + + // HubSyncWrite is the format for a failed entry write after + // a successful pull. Takes (count, error). + HubSyncWrite = "hubsync: write %d entries: %v" +) + // Warn context identifiers for index generation. const ( // IndexHeader is the context label for index header write errors. diff --git a/internal/hub/eof.go b/internal/hub/eof.go index e3bf4a70a..081c305ea 100644 --- a/internal/hub/eof.go +++ b/internal/hub/eof.go @@ -6,15 +6,27 @@ package hub -import stdio "io" +import ( + stdio "io" +) -// eof reports whether err is io.EOF. +// eof reports whether err is exactly io.EOF. +// +// The equality is deliberately strict. eof gates the clean +// end-of-stream path in client.Sync and client.Listen, where a true +// result means "the stream ended normally, treat what we have as +// complete." gRPC delivers a bare, unwrapped io.EOF only on an OK +// status, so == is correct today; keeping it strict means that if a +// future layer ever wraps a mid-stream (dirty) EOF, it is treated as +// a real error rather than silently accepted as a clean end. The +// replication warn-suppression path wants the opposite polarity and +// broadens to errors.Is inline at its own call site. // // Parameters: // - err: error to check // // Returns: -// - bool: true if err is io.EOF +// - bool: true if err is exactly io.EOF func eof(err error) bool { return err == stdio.EOF } diff --git a/internal/hub/replicate.go b/internal/hub/replicate.go index d8475fdae..8f60f95b0 100644 --- a/internal/hub/replicate.go +++ b/internal/hub/replicate.go @@ -8,6 +8,8 @@ package hub import ( "context" + "errors" + stdio "io" "time" cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" @@ -81,6 +83,7 @@ func replicateOnce( ), ) if dialErr != nil { + logWarn.Warn(cfgWarn.HubReplicateDial, masterAddr, dialErr) return } defer func() { @@ -98,21 +101,38 @@ func replicateOnce( cfgHub.PathSync, ) if streamErr != nil { + logWarn.Warn(cfgWarn.HubReplicateStream, masterAddr, streamErr) return } if sendErr := stream.SendMsg(&SyncRequest{ SinceSequence: lastSeq, }); sendErr != nil { + logWarn.Warn(cfgWarn.HubReplicateSend, masterAddr, sendErr) return } if closeErr := stream.CloseSend(); closeErr != nil { + logWarn.Warn(cfgWarn.HubReplicateCloseSend, masterAddr, closeErr) return } for { msg := &EntryMsg{} if recvErr := stream.RecvMsg(msg); recvErr != nil { + // io.EOF is the normal end of every sync stream + // and a done caller context is routine shutdown; + // warning on either would spam stderr once per + // replication cycle. Anything else is a transport + // failure worth surfacing. errors.Is is used inline + // (rather than the strict eof helper) so a wrapped EOF + // is still suppressed here — the lenient polarity this + // warn-suppression path wants, opposite to eof's + // strict clean-end checks in client.go. + if !errors.Is(recvErr, stdio.EOF) && ctx.Err() == nil { + logWarn.Warn( + cfgWarn.HubReplicateRecv, masterAddr, recvErr, + ) + } return } entry := Entry{ diff --git a/internal/hub/replicate_test.go b/internal/hub/replicate_test.go new file mode 100644 index 000000000..dda3d0ed0 --- /dev/null +++ b/internal/hub/replicate_test.go @@ -0,0 +1,203 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package hub + +import ( + "bytes" + "net" + "os" + "runtime" + "strings" + "testing" + "time" + + "github.com/ActiveMemory/ctx/internal/config/fs" + logWarn "github.com/ActiveMemory/ctx/internal/log/warn" +) + +// captureWarnings redirects the warn sink to a buffer for the +// duration of the test. +func captureWarnings(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + restore := logWarn.SetSink(&buf) + t.Cleanup(restore) + return &buf +} + +// startMaster serves a hub over a fresh store on a random port +// and returns the store, the address, and the admin token. +func startMaster(t *testing.T) (*Store, string, string) { + t.Helper() + store, storeErr := NewStore(t.TempDir()) + if storeErr != nil { + t.Fatal(storeErr) + } + adminTok, tokErr := GenerateAdminToken() + if tokErr != nil { + t.Fatal(tokErr) + } + srv := NewServer(store, adminTok) + lis := listenRandom(t) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.GracefulStop) + return store, lis.Addr().String(), adminTok +} + +// registerClient registers a project and returns its token. +func registerClient(t *testing.T, addr, adminTok string) string { + t.Helper() + client, dialErr := NewClient(addr, "") + if dialErr != nil { + t.Fatal(dialErr) + } + defer func() { + if cerr := client.Close(); cerr != nil { + t.Log(cerr) + } + }() + reg, regErr := client.Register( + testCtx(), adminTok, "replicate-test", + ) + if regErr != nil { + t.Fatal(regErr) + } + return reg.ClientToken +} + +// seedEntries appends n entries to the master store. +func seedEntries(t *testing.T, store *Store, n int) { + t.Helper() + entries := make([]Entry, n) + for i := range entries { + entries[i] = Entry{ + ID: "e" + string(rune('a'+i)), + Type: "decision", + Content: "replicated content", + Origin: "replicate-test", + Timestamp: time.Now(), + } + } + if _, appendErr := store.Append(entries); appendErr != nil { + t.Fatal(appendErr) + } +} + +func TestReplicateOnce_WarnsOnDialError(t *testing.T) { + buf := captureWarnings(t) + follower, storeErr := NewStore(t.TempDir()) + if storeErr != nil { + t.Fatal(storeErr) + } + + // grpc.NewClient is lazy for almost every bad target, but + // a control character fails URL parsing at construction — + // the one eager failure mode, and exactly what a corrupted + // peer config would produce. + replicateOnce(testCtx(), "\x00", follower, "tok") + + if !strings.Contains(buf.String(), "hub replicate dial") { + t.Errorf("missing dial warning, got: %q", buf.String()) + } +} + +func TestReplicateOnce_WarnsOnTransportError(t *testing.T) { + buf := captureWarnings(t) + follower, storeErr := NewStore(t.TempDir()) + if storeErr != nil { + t.Fatal(storeErr) + } + + // A well-formed address nobody listens on. Client and + // stream construction are lazy, so the connection failure + // surfaces at whichever stage first touches the wire — + // any of the replication warnings is correct. + lis, lisErr := net.Listen("tcp", "127.0.0.1:0") + if lisErr != nil { + t.Fatal(lisErr) + } + addr := lis.Addr().String() + if closeErr := lis.Close(); closeErr != nil { + t.Fatal(closeErr) + } + + replicateOnce(testCtx(), addr, follower, "tok") + + if !strings.Contains(buf.String(), "hub replicate") { + t.Errorf( + "missing transport warning, got: %q", buf.String(), + ) + } +} + +func TestReplicateOnce_CleanReplicationDoesNotWarn(t *testing.T) { + master, addr, adminTok := startMaster(t) + token := registerClient(t, addr, adminTok) + seedEntries(t, master, 2) + + follower, storeErr := NewStore(t.TempDir()) + if storeErr != nil { + t.Fatal(storeErr) + } + buf := captureWarnings(t) + + replicateOnce(testCtx(), addr, follower, token) + + if buf.Len() != 0 { + t.Errorf( + "clean replication must not warn, got: %q", + buf.String(), + ) + } + if _, lastSeq := follower.lastSequence(); lastSeq != 2 { + t.Errorf("follower lastSeq = %d, want 2", lastSeq) + } +} + +func TestReplicateOnce_KeepsConsumingAfterAppendError(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("permission semantics differ on windows") + } + if os.Geteuid() == 0 { + t.Skip("root bypasses permission checks") + } + master, addr, adminTok := startMaster(t) + token := registerClient(t, addr, adminTok) + seedEntries(t, master, 2) + + followerDir := t.TempDir() + follower, storeErr := NewStore(followerDir) + if storeErr != nil { + t.Fatal(storeErr) + } + // Make every append fail: the store opens its files per + // call, so an access-denied directory rejects each write. + if chmodErr := os.Chmod(followerDir, 0); chmodErr != nil { + t.Fatal(chmodErr) + } + t.Cleanup(func() { + if chmodErr := os.Chmod( + followerDir, fs.PermExec, + ); chmodErr != nil { + t.Log(chmodErr) + } + }) + buf := captureWarnings(t) + + replicateOnce(testCtx(), addr, follower, token) + + // Both entries must have been attempted: an append failure + // is warned per entry and must not abort the stream. + got := strings.Count(buf.String(), "hub replicate append") + if got != 2 { + t.Errorf( + "append warnings = %d, want 2 (loop must keep "+ + "consuming); output: %q", + got, buf.String(), + ) + } +} diff --git a/specs/fix-hub-silent-error-suppression.md b/specs/fix-hub-silent-error-suppression.md new file mode 100644 index 000000000..2390df8a3 --- /dev/null +++ b/specs/fix-hub-silent-error-suppression.md @@ -0,0 +1,160 @@ +# Fix Hub Silent Error Suppression + +The session-start hubsync hook and the cluster replication +loop swallowed errors with no logging surface: operators could +not tell whether a sync succeeded, partially failed, or never +reached the network. Upstream issue: +[ActiveMemory/ctx#100](https://github.com/ActiveMemory/ctx/issues/100). + +## Problem + +### Hubsync hook — `internal/cli/system/core/hubsync/sync.go` + +`Sync` returned `""` silently on config-load failure, dial +failure, sync-RPC failure, and entry-write failure. Worse, the +sync-error check was conflated with the empty-result check: + +```go +entries, syncErr := client.Sync( + context.Background(), cfg.Types, 0, +) +if syncErr != nil || len(entries) == 0 { + return "" +} +``` + +A real network error was indistinguishable from "nothing new." +The package doc codified the behavior ("Every error is +silently swallowed so the hook never blocks the session +start"). The never-block constraint is correct; the silence is +the bug. + +### Replication loop — `internal/hub/replicate.go` + +`replicateOnce` returned silently on dial, stream-open, send, +and close-send failures, and on every receive error — including +real transport failures. (The `conn.Close` defer and the +`store.Append` failure path already warn, and append already +keeps consuming the stream; those two sub-items of #100 landed +upstream before this change.) + +## Solution + +Wire every silent return through the established +`internal/log/warn` sink with format constants in +`internal/config/warn`, preserving both functions' signatures +and non-blocking contracts. Logging is the only behavior +change, plus one un-conflation: + +1. `internal/config/warn/warn.go` — nine new format + constants: `HubSyncLoadConfig`, `HubSyncDial`, + `HubSyncPull`, `HubSyncWrite` (hubsync hook; `hubsync:` + prefix per the `notify:` precedent) and + `HubReplicateDial`, `HubReplicateStream`, + `HubReplicateSend`, `HubReplicateCloseSend`, + `HubReplicateRecv` (extending the existing + `HubReplicateAppend` family). +2. `internal/cli/system/core/hubsync/sync.go` — warn at all + four silent sites; split `syncErr` from the + `len(entries) == 0` check so only the error case warns. A + genuine empty result stays silent. +3. `internal/cli/system/core/hubsync/doc.go` — the contract + sentence becomes "Every error is surfaced as a stderr + warning via the warn sink, but never propagates: the hook + must not block the session start." +4. `internal/hub/replicate.go` — warn at dial, stream-open, + send, and close-send failures. The receive site + distinguishes three cases: `io.EOF` is the normal end of + every sync stream (returns silently — warning here would + spam stderr once per `ReplicateInterval`); a done caller + context is routine shutdown noise (silent); anything else + is a transport failure and warns. Issue #100's proposed + code warns on every receive error and would have made + clean replication cycles noisy; this is the one deliberate + deviation. + +## Tests + +`warn.SetSink` (existing test seam) captures output in all of +them. + +- `internal/cli/system/core/hubsync/sync_test.go` (new; the + package had no tests): + - `TestSync_WarnsOnLoadError` — no connect config present; + warns `hubsync: load connection config:`. + - `TestSync_WarnsOnDialError` — `HubAddr` containing a + control character. Empirically the only eager + `grpc.NewClient` failure mode: almost every malformed + target (`://invalid`, `unix://not-abs`) is deferred to + first use by the lazy resolver, but a control character + fails URL parsing at construction. Warns `hubsync: dial`. + - `TestSync_WarnsOnPullError` — well-formed but closed + address; `grpc.NewClient` is lazy, so the failure + surfaces at the Sync RPC; warns `hubsync: sync from`. + - `TestSync_NoWarnOnEmptyResult` — real in-process hub + with zero entries; no warning, empty return (pins the + un-conflation). +- `internal/hub/replicate_test.go` (new; `replicateOnce` had + no direct coverage): + - `TestReplicateOnce_WarnsOnDialError` — master target + with a control character (same eager-failure rationale + as the hubsync dial test). + - `TestReplicateOnce_WarnsOnTransportError` — closed port; + asserts a `hub replicate` warning from whichever lazy + stage surfaces the failure. + - `TestReplicateOnce_CleanReplicationDoesNotWarn` — real + master with two entries, writable follower; entries + replicate, `io.EOF` ends the cycle, zero warnings (pins + the EOF deviation). + - `TestReplicateOnce_KeepsConsumingAfterAppendError` — + read-only follower store directory; both appends fail, + two `hub replicate append` warnings, loop reaches EOF + (pins continue-on-append-failure). + +## Review Follow-ups (PR #114) + +Code review surfaced that the re-documented "must not block the +session start" contract was not actually guaranteed, plus two +smaller hardening points. Addressed here: + +1. **Bounded pull deadline.** `Sync` called the RPC with + `context.Background()` and no deadline. `grpc.NewClient` is + lazy and fail-fast (not `WaitForReady`), but a hub that + accepts the TCP connection and then never responds (hung + server, black-hole proxy) makes the underlying `RecvMsg` + block indefinitely — so the hook could hang despite the + contract. The pull now runs under + `context.WithTimeout(..., HubSyncTimeout)` (new constant in + `internal/config/hub`, 10s). An exceeded deadline surfaces + through the existing `HubSyncPull` warning and the hook + returns `""` like any other failure; the daily throttle plus + next-session retry covers a cut-off pull. The timeout is a + package var (`syncTimeout`) so tests can shrink it; a new + `TestSync_WarnsOnHungHub` drives a black-hole listener and + asserts `Sync` returns (rather than hanging) with the pull + warning. +2. **`eof()` uses `errors.Is`.** `internal/hub/eof.go` compared + with `==`. gRPC delivers an unwrapped `io.EOF` today (the + clean-replication test confirms it), but `errors.Is` keeps + the EOF suppression correct should any layer ever wrap it, + rather than leaking a wrapped EOF as a per-cycle transport + warning. Pre-existing helper; this PR newly relies on it for + the receive-suppression branch. + +## Out of Scope + +- Structured (JSON) event-log emission; stderr via + `warn.Warn` is the established pattern (issue's own + out-of-scope list). +- Making `Sync` return an error — the non-blocking hook + contract is hard. +- Wiring `startReplication` into `Server.Start` (#96 + territory) and the hubsync hook's hardcoded + `sinceSequence=0` full-refetch (a separate latent issue, + noted during review of #93). +- Dedup/backoff for the replication receive warning. Once + `startReplication` is wired, a persistently-down-but-reachable + master would emit `HubReplicateRecv` every `ReplicateInterval` + with no rate-limiting. `replicateOnce` is dead code today, so + this is deferred to the wiring work (tracked as a task) rather + than guessed at now.