fix: drive mesh convergence from the replication feed

cmd/meshd/main.go

@@ -1734,6 +1734,10 @@ func cmdNetworkLeave(ctx context.Context, args []string, flagProfile string) err
 	if err := state.ClearNetworkState(stateDir); err != nil {
 		return fmt.Errorf("clearing network state: %w", err)
 	}
+	// The feed checkpoint belongs to the departed network's replication link.
+	if err := state.ClearFeedSyncState(stateDir); err != nil {
+		fmt.Printf("Warning: clearing feed sync state: %v\n", err)
+	}
 
 	name := "unknown"
 	if ns != nil {
@@ -2969,6 +2973,11 @@ func printDaemonRuntimeStatus(w io.Writer, live *daemon.Status) {
 	} else if live.RoutingReady {
 		fmt.Fprintln(w, "  Routing: userspace ready")
 	}
+	if live.Snapshot != nil {
+		if reason := strings.TrimSpace(live.Snapshot.DegradedReason); reason != "" {
+			fmt.Fprintf(w, "  Control plane: degraded — %s\n", reason)
+		}
+	}
 }
 
 type doctorLevel string
@@ -4217,13 +4226,23 @@ func cmdUp(ctx context.Context, args []string, flagProfile string) error {
 		ListenPort:             f.listenPort,
 		PollInterval:           f.pollInterval,
 		Logger:                 logger,
+		// The engine derives feed eligibility (owner signing identity or feed
+		// grants below) and never learns where the checkpoint lives.
+		FeedSyncLoad: func() (*state.FeedSyncState, error) {
+			return state.LoadFeedSyncState(stateDir)
+		},
+		FeedSyncSave: func(fs state.FeedSyncState) error {
+			return state.SaveFeedSyncState(stateDir, &fs)
+		},
 	}
 	if delegateSession != nil {
 		engCfg.DelegatedGrant = delegateSession.ReadGrant
 		engCfg.WriteDelegatedGrant = delegateSession.WriteGrant
 		engCfg.GrantKeys = delegateSession.GrantKeys
 		engCfg.AudienceSource = delegateSession.AudienceSource
 		engCfg.ProtocolDefinition = delegateSession.ProtocolDefinition
+		engCfg.FeedGrantIDs = delegateSession.FeedGrantIDs
+		engCfg.FeedGrants = delegateSession.FeedGrants
 	}
 	socketPath := daemon.DefaultSocketPath()
 	instanceID, err := resolveDaemonInstanceID(backgroundChild, os.Getenv(startupInstanceEnv))
@@ -4352,6 +4371,7 @@ func applyDaemonRefreshHealth(status *daemon.SnapshotStatus, snapshot *engine.Me
 
 	status.State = daemonRefreshState(snapshot, refresh)
 	status.Mode = string(refresh.Mode)
+	status.DegradedReason = refresh.DegradedReason
 	status.InFlight = refresh.InFlight
 	status.Pending = daemonRefreshReasons(refresh.PendingReasons)
 	status.Paused = refresh.Paused

cmd/meshd/peer_list_local_test.go

@@ -258,6 +258,7 @@ func TestDaemonRefreshHealthMapping(t *testing.T) {
 		InFlight:       true,
 		Mode:           engine.RefreshModeFallback,
 		StreamsHealthy: false,
+		DegradedReason: "wallet session grants expired",
 		Streams: map[engine.RefreshStream]engine.RefreshStreamHealth{
 			engine.RefreshStreamTopology: {Covered: true, Live: true, Repaired: true},
 			engine.RefreshStreamDelivery: {Covered: true, Live: false, Repaired: false},
@@ -298,6 +299,36 @@ func TestDaemonRefreshHealthMapping(t *testing.T) {
 		got.Streams["delivery"].Repaired {
 		t.Fatalf("stream status = %+v", got.Streams)
 	}
+	if got.DegradedReason != "wallet session grants expired" {
+		t.Fatalf("degraded reason = %q, want surfaced", got.DegradedReason)
+	}
+}
+
+func TestDaemonRefreshHealthMapsFeedStream(t *testing.T) {
+	health := &engine.RefreshCoordinatorHealth{
+		Running:        true,
+		Mode:           engine.RefreshModeHealthy,
+		StreamsHealthy: true,
+		Streams: map[engine.RefreshStream]engine.RefreshStreamHealth{
+			engine.RefreshStreamFeed: {Covered: true, Live: true, Repaired: true},
+		},
+		LastSuccessAt: time.Date(2026, 7, 11, 12, 0, 0, 0, time.UTC),
+	}
+
+	_, _, got := daemonStatusesFromMeshSnapshot(&engine.MeshSnapshot{Generation: 1}, health)
+	if got == nil {
+		t.Fatal("snapshot status = nil")
+	}
+	if len(got.Streams) != 1 {
+		t.Fatalf("streams = %+v, want only feed", got.Streams)
+	}
+	feed, ok := got.Streams["feed"]
+	if !ok || !feed.Covered || !feed.Live || !feed.Repaired {
+		t.Fatalf("feed stream = %+v (present %v), want covered/live/repaired", feed, ok)
+	}
+	if got.DegradedReason != "" {
+		t.Fatalf("degraded reason = %q, want empty", got.DegradedReason)
+	}
 }
 
 func TestDaemonRefreshState(t *testing.T) {

internal/control/dwnclient.go

@@ -254,6 +254,28 @@ func NewDWNClient(
 	}
 }
 
+// FeedFingerprint fetches the anchor tenant's replication-feed fingerprint
+// for the mesh protocol scope with a cheap cids-only, single-entry
+// MessagesQuery. permissionGrantIDs is the canonical feed grant set of a
+// wallet delegate; the anchor owner passes nil (tenant-authored Messages
+// operations bypass grants).
+//
+// An empty fingerprint with a nil error means the server did not provide one
+// for this filter set; callers must treat that as unknown and perform a full
+// state load.
+func (c *DWNClient) FeedFingerprint(ctx context.Context, permissionGrantIDs []string) (string, error) {
+	result, err := c.anchorDWN.MessagesQuery(ctx, c.anchorTenant, dwn.MessagesQueryOptions{
+		Filters:            []dwn.MessagesFilter{{Protocol: protocols.MeshProtocolURI}},
+		PermissionGrantIDs: permissionGrantIDs,
+		Limit:              1,
+		CidsOnly:           true,
+	})
+	if err != nil {
+		return "", fmt.Errorf("querying feed fingerprint: %w", err)
+	}
+	return result.Fingerprint, nil
+}
+
 // UndecryptablePeerCount returns the cumulative number of node records this
 // client has failed to decrypt since it was created. A non-zero, growing value
 // means role-audience keys are not reaching this node (issue #187): its peers

internal/daemon/daemon.go

@@ -100,6 +100,11 @@ type SnapshotStatus struct {
 	RetryNotBefore      string                          `json:"retryNotBefore,omitempty"`
 	NextAttemptAt       string                          `json:"nextAttemptAt,omitempty"`
 	Streams             map[string]SnapshotStreamStatus `json:"streams,omitempty"`
+
+	// DegradedReason explains a terminal loss of subscription coverage in
+	// operator terms (e.g. wallet session grants expired). Empty while
+	// coverage is intact.
+	DegradedReason string `json:"degradedReason,omitempty"`
 }
 
 // SnapshotStreamStatus describes whether one invalidation stream is

internal/dwn/messages_subscribe_test.go

@@ -260,7 +260,7 @@ func TestMessagesSubscribeTerminalDeliveryErrorCodes(t *testing.T) {
 		wantTerminal bool
 	}{
 		{code: MessagesSubscribeDeliveryAuthorizationFailedCode, wantRevoked: true, wantTerminal: true},
-		{code: MessagesSubscribeDeliveryFailedCode, wantRevoked: true, wantTerminal: true},
+		{code: MessagesSubscribeDeliveryFailedCode, wantRevoked: false, wantTerminal: true},
 		{code: "SomeOtherTerminalCode", wantRevoked: false, wantTerminal: true},
 	}
 

internal/dwn/subscribe.go

@@ -50,6 +50,13 @@ func (p *ProgressToken) valid() bool {
 	return ok
 }
 
+// Valid reports whether the token carries the stream, epoch, and decimal
+// position required to resume a subscription. Callers restoring persisted
+// cursors should discard invalid tokens instead of failing the subscribe.
+func (p *ProgressToken) Valid() bool {
+	return p.valid()
+}
+
 func shouldReplaceProgressToken(current, candidate *ProgressToken) bool {
 	if !candidate.valid() {
 		return false
@@ -124,10 +131,10 @@ const (
 )
 
 // ErrSubscriptionAuthorizationRevoked marks a terminal subscription error
-// caused by the server permanently stopping delivery — grant revocation,
-// grant expiry, or an unrecoverable delivery failure. The terminal lifecycle
-// error wraps this sentinel so callers can distinguish "re-resolve the grant
-// set" from other terminal conditions via errors.Is.
+// caused by delivery-time grant revalidation failing — grant revocation or
+// grant expiry. The terminal lifecycle error wraps this sentinel so callers
+// can distinguish "re-resolve the grant set" from other terminal conditions
+// (including generic delivery failures) via errors.Is.
 var ErrSubscriptionAuthorizationRevoked = errors.New("subscription delivery authorization revoked")
 
 // ProgressGapInfo describes why a stored cursor cannot be resumed.
@@ -797,9 +804,10 @@ func (s *Subscription) handleSubscriptionFrame(ctx context.Context, conn *websoc
 	var terminalErr error
 	if messageType == SubscriptionErrorType {
 		switch message.Error.Code {
-		case MessagesSubscribeDeliveryAuthorizationFailedCode, MessagesSubscribeDeliveryFailedCode:
-			// Grant revocation/expiry (or unrecoverable delivery failure):
-			// wrap the sentinel so callers can react specifically.
+		case MessagesSubscribeDeliveryAuthorizationFailedCode:
+			// Grant revocation/expiry: wrap the sentinel so callers can react
+			// specifically. DeliveryFailed is a server-side delivery fault, not
+			// a grant problem, so it takes the plain terminal path below.
 			terminalErr = fmt.Errorf("terminal subscription error: %s: %s: %w",
 				message.Error.Code, message.Error.Detail, ErrSubscriptionAuthorizationRevoked)
 		default:

internal/engine/dwncontrol.go

@@ -58,13 +58,33 @@ type DWNControlConfig struct {
 	// subscription coverage is incomplete or unhealthy. Default: 30 seconds.
 	PollInterval time.Duration
 
-	// HealthyPollInterval is the slow anti-entropy interval while topology and
-	// delivery streams are both live and repaired. Default: 5 minutes.
+	// HealthyPollInterval is the slow anti-entropy interval while every
+	// required stream is live and repaired. Default: 5 minutes.
 	HealthyPollInterval time.Duration
 
 	// RefreshTimeout bounds one complete DWN state rebuild. Default: 2 minutes.
 	RefreshTimeout time.Duration
 
+	// Streams is the coordinator's authoritative stream set. Nil defaults to
+	// the two Records streams (topology, delivery); feed mode passes [feed].
+	Streams []RefreshStream
+
+	// FeedFingerprintFunc fetches the current remote replication-feed
+	// fingerprint (cids-only MessagesQuery). Non-nil enables fingerprint-gated
+	// periodic refreshes: a periodic-only batch whose fingerprint matches the
+	// one recorded at the last full rebuild skips the full state load.
+	FeedFingerprintFunc func(ctx context.Context) (string, error)
+
+	// FeedSync stores the fingerprint recorded immediately before each
+	// successful full rebuild. Required for gating when FeedFingerprintFunc
+	// is set.
+	FeedSync *feedSyncRecorder
+
+	// FeedFullRefreshInterval bounds how long periodic refreshes may be
+	// fingerprint-skipped before a full rebuild is forced regardless of a
+	// stable fingerprint. Default: 30 minutes.
+	FeedFullRefreshInterval time.Duration
+
 	// StartupSubscriptionWait gives asynchronous subscription handshakes a
 	// bounded chance to establish before the initial full rebuild.
 	StartupSubscriptionWait time.Duration
@@ -134,6 +154,10 @@ type DWNControl struct {
 	cancel            context.CancelFunc
 	coordinator       *RefreshCoordinator
 	initialMapApplied atomic.Bool
+
+	// lastFullRefreshAt is when the last complete state load finished. Only
+	// touched from the coordinator's single refresh worker.
+	lastFullRefreshAt time.Time
 }
 
 // NewDWNControlFactory returns a factory function suitable for
@@ -174,6 +198,10 @@ func NewDWNControl(config *DWNControlConfig, opts controlclient.Options) (*DWNCo
 	if refreshTimeout == 0 {
 		refreshTimeout = 2 * time.Minute
 	}
+	feedFullRefreshInterval := config.FeedFullRefreshInterval
+	if feedFullRefreshInterval == 0 {
+		feedFullRefreshInterval = 30 * time.Minute
+	}
 
 	p := opts.Persist.Clone()
 	if p == nil {
@@ -202,6 +230,10 @@ func NewDWNControl(config *DWNControlConfig, opts controlclient.Options) (*DWNCo
 			StartupSubscriptionWait: config.StartupSubscriptionWait,
 			NodePrivateKey:          config.NodePrivateKey,
 			DiscoKeyRegistry:        config.DiscoKeyRegistry,
+			Streams:                 config.Streams,
+			FeedFingerprintFunc:     config.FeedFingerprintFunc,
+			FeedSync:                config.FeedSync,
+			FeedFullRefreshInterval: feedFullRefreshInterval,
 			Logf:                    logf,
 		},
 		observer: opts.Observer,
@@ -212,6 +244,7 @@ func NewDWNControl(config *DWNControlConfig, opts controlclient.Options) (*DWNCo
 	}
 	coordinator, err := NewRefreshCoordinator(RefreshCoordinatorConfig{
 		Refresh:          cc.refreshControlState,
+		Streams:          config.Streams,
 		FallbackInterval: pollInterval,
 		HealthyInterval:  healthyPollInterval,
 		RetryBackoff:     time.Second,
@@ -287,8 +320,7 @@ func (cc *DWNControl) waitForStartupSubscriptions(ctx context.Context, maximum t
 
 func refreshStreamsReadyForStartup(health RefreshCoordinatorHealth) bool {
 	covered := 0
-	for _, stream := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} {
-		state := health.Streams[stream]
+	for _, state := range health.Streams {
 		if !state.Covered {
 			continue
 		}
@@ -303,18 +335,50 @@ func refreshStreamsReadyForStartup(health RefreshCoordinatorHealth) bool {
 // refreshControlState performs one bounded remote rebuild. The first usable
 // map is applied locally a second time after the event bus has observed its peer
 // views; this preserves the startup ordering workaround without a second DWN load.
-func (cc *DWNControl) refreshControlState(ctx context.Context, _ RefreshBatch) error {
+//
+// In feed mode the remote feed fingerprint is fetched immediately before the
+// load: a periodic-only anti-entropy pass whose fingerprint matches the one
+// recorded at the last full rebuild skips the load entirely (the attempt
+// still counts as a success, keeping health fresh). Event-driven, startup,
+// manual, and endpoint reasons always rebuild, and a full rebuild is forced
+// at least every FeedFullRefreshInterval even when the fingerprint is stable.
+func (cc *DWNControl) refreshControlState(ctx context.Context, batch RefreshBatch) error {
 	refreshCtx := ctx
 	cancel := func() {}
 	if cc.config.RefreshTimeout > 0 {
 		refreshCtx, cancel = context.WithTimeout(ctx, cc.config.RefreshTimeout)
 	}
 	defer cancel()
 
+	fingerprint := ""
+	fingerprintKnown := false
+	if cc.config.FeedFingerprintFunc != nil {
+		fp, err := cc.config.FeedFingerprintFunc(refreshCtx)
+		if err != nil {
+			// Probe failures are non-fatal: fall through to the full load.
+			cc.logf("dwn-control: feed fingerprint probe failed: %v", err)
+		} else {
+			fingerprint = fp
+			fingerprintKnown = fp != ""
+		}
+		if fingerprintKnown && cc.canSkipPeriodicRefresh(batch, fingerprint) {
+			cc.logf("dwn-control: feed fingerprint unchanged; skipping periodic rebuild")
+			return nil
+		}
+	}
+
 	replay, err := cc.loadAndPush(refreshCtx)
 	if err != nil {
 		return err
 	}
+	if cc.config.FeedFingerprintFunc != nil {
+		// Record the fingerprint fetched BEFORE the load succeeded: a write
+		// racing the load leaves the stored fingerprint stale, which only
+		// costs one redundant refresh (the safe direction). A failed probe
+		// stores "" so the next periodic pass cannot skip.
+		cc.config.FeedSync.recordFingerprint(fingerprint)
+		cc.lastFullRefreshAt = time.Now()
+	}
 	if replay == nil || cc.observer == nil || !cc.initialMapApplied.CompareAndSwap(false, true) {
 		return nil
 	}
@@ -325,6 +389,28 @@ func (cc *DWNControl) refreshControlState(ctx context.Context, _ RefreshBatch) e
 	return nil
 }
 
+// canSkipPeriodicRefresh reports whether a batch is pure periodic
+// anti-entropy whose remote fingerprint matches the stored one from the last
+// full rebuild, within the forced-rebuild interval.
+func (cc *DWNControl) canSkipPeriodicRefresh(batch RefreshBatch, fingerprint string) bool {
+	if cc.config.FeedSync == nil || len(batch.Reasons) == 0 {
+		return false
+	}
+	for _, reason := range batch.Reasons {
+		if reason != RefreshReasonPeriodic {
+			return false
+		}
+	}
+	stored := cc.config.FeedSync.storedFingerprint()
+	if stored == "" || stored != fingerprint {
+		return false
+	}
+	if cc.lastFullRefreshAt.IsZero() {
+		return false
+	}
+	return time.Since(cc.lastFullRefreshAt) < cc.config.FeedFullRefreshInterval
+}
+
 // loadAndPush reads DWN state, applies it once, and returns an observer-safe
 // clone for the one-time startup replay.
 func (cc *DWNControl) loadAndPush(ctx context.Context) (*netmap.NetworkMap, error) {