feat(swim): add self-refutation and cross-process HTTP gossip dissemination

Implement SWIM self-refute so a node receiving a suspect/dead claim
about itself at incarnation >= local bumps its incarnation and re-marks
Alive, propagating the refutation cluster-wide via higher-incarnation-wins.

Wire cross-process gossip over HTTP:
- Add Gossip(ctx, targetID, members) to DistTransport interface
- Add POST /internal/gossip server endpoint (auth-wrapped)
- Introduce GossipMember wire DTO with projection helpers
- runGossipTick now falls through to the HTTP transport for
  non-InProcessTransport clusters (previously a no-op)

Swap encoding/json for github.com/goccy/go-json in the server binary.

Remove the experimental qualifier from heartbeat/failure-detection
docs — indirect probes (Phase B.1) and self-refutation (Phase E)
together provide the SWIM guarantees the marker was tracking.

Tests: TestDistSWIM_HTTPGossipExchange, TestDistSWIM_SelfRefute

Author: hyp3rd

CHANGELOG.md

@@ -8,6 +8,36 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Added
 
+- **SWIM self-refutation + cross-process gossip dissemination.**
+  Closes the last `experimental` marker on the heartbeat path.
+  Three pieces:
+   - **`acceptGossip` self-refute** — incoming entries that
+     reference the local node as Suspect or Dead at incarnation
+     ≥ ours now bump the local incarnation and re-mark Alive.
+     Higher-incarnation-wins propagation in the same function
+     disseminates the refutation cluster-wide, so a falsely-
+     suspected node can clear suspicion through gossip alone
+     (pre-fix the only path was a fresh probe).
+   - **HTTP gossip wire** — new `Gossip(ctx, targetID, members)`
+     method on `DistTransport`, new
+     `POST /internal/gossip` server endpoint (auth-wrapped),
+     new `GossipMember` wire DTO. `runGossipTick` now falls
+     through to the HTTP path when the transport isn't an
+     `InProcessTransport`, so cross-process clusters disseminate
+     membership state — pre-Phase-E this was an in-process-only
+     no-op.
+   - The `experimental` qualifier is removed from
+     `heartbeatLoop`'s comment + the heartbeat-section field
+     doc; SWIM-style indirect probes (Phase B.1) and
+     self-refutation (this round) together provide the SWIM
+     properties the marker was tracking.
+  Regression coverage at
+  [tests/integration/dist_swim_refute_test.go](tests/integration/dist_swim_refute_test.go):
+  `TestDistSWIM_HTTPGossipExchange` exercises the wire (A pushes
+  membership to B over HTTP; B's view converges),
+  `TestDistSWIM_SelfRefute` drives a forged "you are suspect"
+  gossip into a node's `/internal/gossip` and asserts the local
+  incarnation bumps + state returns to Alive.
 - **End-to-end resilience test** at
   [scripts/tests/20-test-cluster-resilience.sh](scripts/tests/20-test-cluster-resilience.sh)
   — kills a docker container mid-run, asserts the surviving 4

cmd/hypercache-server/main.go

@@ -19,7 +19,6 @@ package main
 import (
 	"context"
 	"encoding/base64"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -31,6 +30,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/goccy/go-json"
 	fiber "github.com/gofiber/fiber/v3"
 
 	"github.com/hyp3rd/hypercache"

cmd/hypercache-server/main_test.go

@@ -1,13 +1,13 @@
 package main
 
 import (
-	"encoding/json"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
 
+	"github.com/goccy/go-json"
 	fiber "github.com/gofiber/fiber/v3"
 )
 

hypercache-server

@@ -354,6 +354,7 @@ func (s *distHTTPServer) start(bindCtx context.Context, dm *DistMemory) error {
 	s.registerHealth(dm)
 	s.registerDrain(dm)
 	s.registerProbe(dm)
+	s.registerGossip(dm)
 	s.registerMerkle(dm)
 
 	return s.listen(bindCtx)
@@ -516,6 +517,30 @@ func (s *distHTTPServer) registerProbe(dm *DistMemory) {
 	}))
 }
 
+// registerGossip wires `POST /internal/gossip` — the SWIM
+// membership-dissemination endpoint. The body is a JSON array of
+// GossipMember snapshots; the receiver's acceptGossip merges them
+// via higher-incarnation-wins and self-refutes if any entry
+// claims this node is suspect or dead.
+//
+// Auth-wrapped like the rest of `/internal/*` because gossip can
+// inject membership state — an unauthenticated peer could mark
+// real nodes as dead by spoofing a high-incarnation snapshot.
+func (s *distHTTPServer) registerGossip(dm *DistMemory) {
+	s.app.Post("/internal/gossip", s.wrapAuth(func(fctx fiber.Ctx) error {
+		var members []GossipMember
+
+		err := json.Unmarshal(fctx.Body(), &members)
+		if err != nil {
+			return fctx.Status(fiber.StatusBadRequest).JSON(fiber.Map{constants.ErrorLabel: err.Error()})
+		}
+
+		dm.acceptGossip(gossipMembersToNodes(members))
+
+		return fctx.SendStatus(fiber.StatusOK)
+	}))
+}
+
 func (s *distHTTPServer) registerMerkle(dm *DistMemory) {
 	s.app.Get("/internal/merkle", s.wrapAuth(func(fctx fiber.Ctx) error {
 		tree := dm.BuildMerkleTree()

pkg/backend/dist_http_server.go

@@ -380,6 +380,47 @@ func (t *DistHTTPTransport) IndirectHealth(ctx context.Context, relayNodeID, tar
 	return nil
 }
 
+// Gossip pushes a member-list snapshot to the target's
+// `/internal/gossip` endpoint. The receiver merges via
+// higher-incarnation-wins and may self-refute if the snapshot
+// claims it's suspect — see acceptGossip + refuteIfSuspected.
+//
+// The body is a JSON array of GossipMember; the wire shape is
+// stable (separate type from cluster.Node) so the cluster
+// package can add internal fields without breaking peers running
+// older binaries.
+func (t *DistHTTPTransport) Gossip(ctx context.Context, targetNodeID string, members []GossipMember) error {
+	payload, err := json.Marshal(members)
+	if err != nil {
+		return ewrap.Wrap(err, "marshal gossip payload")
+	}
+
+	hreq, err := t.newNodeRequest(ctx, http.MethodPost, targetNodeID, "/internal/gossip",
+		nil, bytes.NewReader(payload))
+	if err != nil {
+		return ewrap.Wrap(err, errMsgNewRequest)
+	}
+
+	hreq.Header.Set("Content-Type", "application/json")
+
+	resp, err := t.doTrusted(hreq)
+	if err != nil {
+		return err
+	}
+
+	defer drainBody(t.limitedBody(resp))
+
+	if resp.StatusCode == http.StatusNotFound {
+		return sentinel.ErrBackendNotFound
+	}
+
+	if resp.StatusCode >= statusThreshold {
+		return ewrap.Newf("gossip status %d", resp.StatusCode)
+	}
+
+	return nil
+}
+
 // FetchMerkle retrieves a Merkle tree snapshot from a remote node.
 func (t *DistHTTPTransport) FetchMerkle(ctx context.Context, nodeID string) (*MerkleTree, error) {
 	if t == nil {

pkg/backend/dist_http_transport.go

@@ -75,7 +75,12 @@ type DistMemory struct {
 	nodeID       string
 	seeds        []string // static seed node addresses
 
-	// heartbeat / failure detection (experimental)
+	// heartbeat / failure detection. Phase E added SWIM
+	// self-refutation (refuteIfSuspected) and HTTP gossip
+	// dissemination, retiring the prior "experimental" marker —
+	// the path now disseminates suspect/dead transitions across
+	// the cluster and lets a falsely-accused node bump its
+	// incarnation to clear suspicion.
 	hbInterval     time.Duration
 	hbSuspectAfter time.Duration
 	hbDeadAfter    time.Duration
@@ -3023,19 +3028,32 @@ func (dm *DistMemory) runGossipTick() {
 	}
 
 	target := candidates[idxBig.Int64()]
+	transport := dm.loadTransport()
+	snapshot := dm.membership.List()
 
-	ip, ok := dm.loadTransport().(*InProcessTransport)
-	if !ok {
-		return
-	}
+	// In-process fast path: skip the wire and call acceptGossip
+	// directly. Pre-Phase-E this was the ONLY path; the function
+	// bailed for any other transport type, so cross-process
+	// clusters never disseminated membership / never refuted
+	// suspect claims. The fall-through below now uses the
+	// transport's Gossip method, which routes via HTTP for the
+	// auto-created DistHTTPTransport.
+	if ip, ok := transport.(*InProcessTransport); ok {
+		if remote, ok2 := ip.backends[string(target.ID)]; ok2 {
+			remote.acceptGossip(snapshot)
+		}
 
-	remote, ok2 := ip.backends[string(target.ID)]
-	if !ok2 {
 		return
 	}
 
-	snapshot := dm.membership.List()
-	remote.acceptGossip(snapshot)
+	gossipErr := transport.Gossip(dm.lifeCtx, string(target.ID), nodesToGossipMembers(snapshot))
+	if gossipErr != nil {
+		dm.logger.Debug(
+			"gossip push failed",
+			slog.String("peer_id", string(target.ID)),
+			slog.Any("err", gossipErr),
+		)
+	}
 }
 
 func (dm *DistMemory) acceptGossip(nodes []*cluster.Node) {
@@ -3045,6 +3063,8 @@ func (dm *DistMemory) acceptGossip(nodes []*cluster.Node) {
 
 	for _, node := range nodes {
 		if node.ID == dm.localNode.ID {
+			dm.refuteIfSuspected(node)
+
 			continue
 		}
 
@@ -3079,6 +3099,41 @@ func (dm *DistMemory) acceptGossip(nodes []*cluster.Node) {
 	}
 }
 
+// refuteIfSuspected handles the SWIM self-refute path: when a peer
+// gossips that THIS node is Suspect or Dead at incarnation N, bump
+// our local incarnation to N+1 and re-upsert ourselves as Alive.
+// Higher-incarnation-wins propagation in `acceptGossip` ensures the
+// next gossip tick disseminates the refutation cluster-wide.
+//
+// Pre-fix this path was a no-op (`continue` on local-ID match) — a
+// node that fell briefly behind heartbeat would be marked Suspect by
+// peers and could not undo it through gossip; only a fresh probe
+// would clear suspicion. Self-refute closes the loop required for
+// the heartbeat marker to drop its `experimental` qualifier.
+func (dm *DistMemory) refuteIfSuspected(claim *cluster.Node) {
+	if claim == nil || dm.localNode == nil {
+		return
+	}
+
+	if claim.State == cluster.NodeAlive {
+		return // peer agrees we're alive — nothing to refute
+	}
+
+	// Only refute when the peer's claim is at >= our incarnation;
+	// older claims are stale and ignored.
+	if claim.Incarnation < dm.localNode.Incarnation {
+		return
+	}
+
+	dm.membership.Mark(dm.localNode.ID, cluster.NodeAlive)
+
+	dm.logger.Info(
+		"self-refuted suspect/dead claim from peer",
+		slog.Uint64("claim_incarnation", claim.Incarnation),
+		slog.String("claim_state", claim.State.String()),
+	)
+}
+
 // chooseNewer picks the item with higher version; on version tie uses lexicographically smaller Origin as winner.
 func (dm *DistMemory) chooseNewer(itemA, itemB *cache.Item) *cache.Item {
 	if itemA == nil {
@@ -3265,7 +3320,10 @@ func parseSeedSpec(raw string) seedSpec {
 	return seedSpec{id: id, addr: addr}
 }
 
-// heartbeatLoop probes peers and updates membership (best-effort experimental).
+// heartbeatLoop probes peers and updates membership. SWIM-style
+// indirect probes (Phase B.1) and self-refutation via gossip
+// (Phase E) are wired into the surrounding helpers — this loop
+// only schedules the per-tick work.
 func (dm *DistMemory) heartbeatLoop(ctx context.Context, stopCh <-chan struct{}) { // reduced cognitive complexity via helpers
 	ticker := time.NewTicker(dm.hbInterval)
 	defer ticker.Stop()

pkg/backend/dist_memory.go

@@ -3,7 +3,9 @@ package backend
 import (
 	"context"
 	"sync"
+	"time"
 
+	"github.com/hyp3rd/hypercache/internal/cluster"
 	"github.com/hyp3rd/hypercache/internal/sentinel"
 	cache "github.com/hyp3rd/hypercache/pkg/cache/v2"
 )
@@ -21,9 +23,87 @@ type DistTransport interface {
 	// the caller's local network was the issue, not the target.
 	// Returns nil when the relay reports the target reachable.
 	IndirectHealth(ctx context.Context, relayNodeID, targetNodeID string) error
+	// Gossip pushes the caller's full member-list snapshot to
+	// `targetNodeID`. The receiver merges it via higher-incarnation-
+	// wins and self-refutes if the snapshot claims it is suspect.
+	// Used by the cross-process gossip path; in-process clusters
+	// short-circuit to a direct method call instead.
+	Gossip(ctx context.Context, targetNodeID string, members []GossipMember) error
 	FetchMerkle(ctx context.Context, nodeID string) (*MerkleTree, error)
 }
 
+// GossipMember is the wire-friendly snapshot of a cluster.Node used
+// by the Gossip transport method. Stays a separate struct from
+// cluster.Node so the wire schema doesn't drift when the cluster
+// package adds internal fields.
+type GossipMember struct {
+	ID          string `json:"id"`
+	Address     string `json:"address"`
+	State       string `json:"state"`
+	Incarnation uint64 `json:"incarnation"`
+}
+
+// nodesToGossipMembers projects a cluster.Node snapshot down to the
+// wire shape. Nil entries are dropped — they shouldn't appear in
+// practice but the projection is defensive.
+func nodesToGossipMembers(nodes []*cluster.Node) []GossipMember {
+	out := make([]GossipMember, 0, len(nodes))
+
+	for _, n := range nodes {
+		if n == nil {
+			continue
+		}
+
+		out = append(out, GossipMember{
+			ID:          string(n.ID),
+			Address:     n.Address,
+			State:       n.State.String(),
+			Incarnation: n.Incarnation,
+		})
+	}
+
+	return out
+}
+
+// gossipMembersToNodes inflates a wire-shape snapshot back into
+// cluster.Node values for handoff to acceptGossip. Unknown state
+// strings fall back to NodeAlive — the receiver's
+// higher-incarnation-wins logic still applies, and a stuck-suspect
+// claim from a peer running an older state vocabulary degrades
+// gracefully to alive-at-this-incarnation.
+func gossipMembersToNodes(members []GossipMember) []*cluster.Node {
+	out := make([]*cluster.Node, 0, len(members))
+
+	for _, m := range members {
+		out = append(out, &cluster.Node{
+			ID:          cluster.NodeID(m.ID),
+			Address:     m.Address,
+			State:       parseGossipState(m.State),
+			Incarnation: m.Incarnation,
+			LastSeen:    time.Now(),
+		})
+	}
+
+	return out
+}
+
+// parseGossipState maps the wire state string back to the
+// internal NodeState enum. "alive" and unknown values both
+// resolve to NodeAlive (defensive — see gossipMembersToNodes);
+// the explicit "alive" branch is omitted to satisfy the
+// identical-switch-branches lint while keeping the same
+// semantic.
+func parseGossipState(s string) cluster.NodeState {
+	switch s {
+	case "suspect":
+		return cluster.NodeSuspect
+	case "dead":
+		return cluster.NodeDead
+	default:
+		return cluster.NodeAlive
+	}
+}
+
 // InProcessTransport implements DistTransport for multiple DistMemory instances in the same process.
 type InProcessTransport struct {
 	mu       sync.RWMutex
@@ -119,6 +199,22 @@ func (t *InProcessTransport) IndirectHealth(ctx context.Context, relayNodeID, ta
 	return rt.Health(ctx, targetNodeID)
 }
 
+// Gossip delivers the snapshot directly to the target backend's
+// acceptGossip — this is the in-process equivalent of the HTTP
+// `/internal/gossip` endpoint, with the type translation done
+// inline so the rest of the SWIM machinery can stay agnostic to
+// transport choice.
+func (t *InProcessTransport) Gossip(_ context.Context, targetNodeID string, members []GossipMember) error {
+	target, ok := t.lookup(targetNodeID)
+	if !ok {
+		return sentinel.ErrBackendNotFound
+	}
+
+	target.acceptGossip(gossipMembersToNodes(members))
+
+	return nil
+}
+
 // FetchMerkle fetches a remote merkle tree.
 func (t *InProcessTransport) FetchMerkle(_ context.Context, nodeID string) (*MerkleTree, error) {
 	b, ok := t.lookup(nodeID)