From 5a7039cacc939acc2cec802976f7bc0e22a4c0ee Mon Sep 17 00:00:00 2001 From: Liran Cohen Date: Sun, 26 Jul 2026 15:53:50 +0000 Subject: [PATCH 1/4] fix(state): persist replication-feed checkpoints bound to a link key Add feedsync.json ({version, linkKey, cursor, fingerprint, updatedAt}) to the profile state dir with the same atomic-write + alignOwnerToDir rules as network.json, so a root daemon leaves it readable by the non-root CLI. Checkpoints from an incompatible schema version are discarded on load, and clearing is wired for network leave. Export ProgressToken.Valid so restorers can discard unusable persisted cursors instead of failing the subscribe. Refs #267 Co-Authored-By: Claude Fable 5 --- internal/dwn/subscribe.go | 7 ++ internal/state/feedsync.go | 100 +++++++++++++++++++++++++ internal/state/feedsync_test.go | 126 ++++++++++++++++++++++++++++++++ 3 files changed, 233 insertions(+) create mode 100644 internal/state/feedsync.go create mode 100644 internal/state/feedsync_test.go diff --git a/internal/dwn/subscribe.go b/internal/dwn/subscribe.go index 212556a..0d22f4e 100644 --- a/internal/dwn/subscribe.go +++ b/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 diff --git a/internal/state/feedsync.go b/internal/state/feedsync.go new file mode 100644 index 0000000..4e8ff39 --- /dev/null +++ b/internal/state/feedsync.go @@ -0,0 +1,100 @@ +package state + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/enboxorg/meshd/internal/dwn" +) + +// FeedSyncStateVersion identifies the feedsync.json schema. Files written by +// an incompatible version are discarded on load. +const FeedSyncStateVersion = 1 + +const feedSyncFile = "feedsync.json" + +// FeedSyncState is the persisted replication-feed checkpoint of the daemon's +// Messages feed subscription: the resume cursor plus the remote feed +// fingerprint recorded at the last successful full state load. +// +// LinkKey binds the checkpoint to one replication link identity +// (tenant^endpoint^projectionId^authorizationEpoch). Grant churn or an +// endpoint change produces a different link key, so a stale checkpoint is +// discarded instead of resumed against the wrong stream authorization. +type FeedSyncState struct { + Version int `json:"version"` + LinkKey string `json:"linkKey"` + Cursor *dwn.ProgressToken `json:"cursor,omitempty"` + Fingerprint string `json:"fingerprint,omitempty"` + UpdatedAt string `json:"updatedAt,omitempty"` +} + +// SaveFeedSyncState persists the feed checkpoint atomically, stamping Version +// and UpdatedAt. +func SaveFeedSyncState(stateDir string, fs *FeedSyncState) error { + if fs == nil { + return fmt.Errorf("feed sync state is required") + } + if fs.LinkKey == "" { + return fmt.Errorf("feed sync state requires a link key") + } + if err := os.MkdirAll(stateDir, 0700); err != nil { + return fmt.Errorf("create state dir: %w", err) + } + fs.Version = FeedSyncStateVersion + fs.UpdatedAt = time.Now().UTC().Format(time.RFC3339Nano) + + data, err := json.MarshalIndent(fs, "", " ") + if err != nil { + return fmt.Errorf("marshal: %w", err) + } + + target := filepath.Join(stateDir, feedSyncFile) + tmp := target + ".tmp" + if err := os.WriteFile(tmp, data, 0600); err != nil { + return fmt.Errorf("write: %w", err) + } + // Same sudo TUN handoff rule as network.json: a root daemon must leave + // the file readable by the profile owner's non-root CLI. + alignOwnerToDir(stateDir, tmp) + if err := os.Rename(tmp, target); err != nil { + os.Remove(tmp) + return fmt.Errorf("rename: %w", err) + } + return nil +} + +// LoadFeedSyncState loads the persisted feed checkpoint. It returns nil, nil +// when no checkpoint exists or the file was written by an incompatible +// version. +func LoadFeedSyncState(stateDir string) (*FeedSyncState, error) { + data, err := os.ReadFile(filepath.Join(stateDir, feedSyncFile)) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, fmt.Errorf("read: %w", err) + } + + var fs FeedSyncState + if err := json.Unmarshal(data, &fs); err != nil { + return nil, fmt.Errorf("unmarshal: %w", err) + } + if fs.Version != FeedSyncStateVersion { + return nil, nil + } + return &fs, nil +} + +// ClearFeedSyncState removes the persisted feed checkpoint (for network +// leave). Missing files are not an error. +func ClearFeedSyncState(stateDir string) error { + err := os.Remove(filepath.Join(stateDir, feedSyncFile)) + if err != nil && !os.IsNotExist(err) { + return err + } + return nil +} diff --git a/internal/state/feedsync_test.go b/internal/state/feedsync_test.go new file mode 100644 index 0000000..5e885a6 --- /dev/null +++ b/internal/state/feedsync_test.go @@ -0,0 +1,126 @@ +package state + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/enboxorg/meshd/internal/dwn" +) + +func TestFeedSyncStateRoundTrip(t *testing.T) { + dir := t.TempDir() + + saved := &FeedSyncState{ + LinkKey: "did:dht:owner^https://dwn.example^proj^epoch", + Cursor: &dwn.ProgressToken{ + StreamID: "aabbccddeeff0011", + Epoch: "epoch-1", + Position: "42", + MessageCID: "bafyreib", + }, + Fingerprint: "0f" + "00", + } + if err := SaveFeedSyncState(dir, saved); err != nil { + t.Fatalf("SaveFeedSyncState: %v", err) + } + if saved.Version != FeedSyncStateVersion { + t.Fatalf("saved version = %d, want %d", saved.Version, FeedSyncStateVersion) + } + if saved.UpdatedAt == "" { + t.Fatal("saved UpdatedAt is empty") + } + + loaded, err := LoadFeedSyncState(dir) + if err != nil { + t.Fatalf("LoadFeedSyncState: %v", err) + } + if loaded == nil { + t.Fatal("loaded state is nil") + } + if loaded.LinkKey != saved.LinkKey || loaded.Fingerprint != saved.Fingerprint { + t.Fatalf("loaded = %+v, want link key %q fingerprint %q", loaded, saved.LinkKey, saved.Fingerprint) + } + if loaded.Cursor == nil || *loaded.Cursor != *saved.Cursor { + t.Fatalf("loaded cursor = %+v, want %+v", loaded.Cursor, saved.Cursor) + } + + // A cleared cursor persists as an absent field. + saved.Cursor = nil + if err := SaveFeedSyncState(dir, saved); err != nil { + t.Fatalf("SaveFeedSyncState without cursor: %v", err) + } + loaded, err = LoadFeedSyncState(dir) + if err != nil { + t.Fatalf("LoadFeedSyncState: %v", err) + } + if loaded.Cursor != nil { + t.Fatalf("loaded cursor = %+v, want nil", loaded.Cursor) + } +} + +func TestLoadFeedSyncStateMissing(t *testing.T) { + loaded, err := LoadFeedSyncState(t.TempDir()) + if err != nil { + t.Fatalf("LoadFeedSyncState: %v", err) + } + if loaded != nil { + t.Fatalf("loaded = %+v, want nil for a missing file", loaded) + } +} + +func TestLoadFeedSyncStateDiscardsIncompatibleVersion(t *testing.T) { + dir := t.TempDir() + data, err := json.Marshal(FeedSyncState{Version: FeedSyncStateVersion + 1, LinkKey: "link"}) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, feedSyncFile), data, 0600); err != nil { + t.Fatal(err) + } + + loaded, err := LoadFeedSyncState(dir) + if err != nil { + t.Fatalf("LoadFeedSyncState: %v", err) + } + if loaded != nil { + t.Fatalf("loaded = %+v, want nil for an incompatible version", loaded) + } +} + +func TestLoadFeedSyncStateCorruptFileErrors(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, feedSyncFile), []byte("{"), 0600); err != nil { + t.Fatal(err) + } + if _, err := LoadFeedSyncState(dir); err == nil { + t.Fatal("LoadFeedSyncState succeeded on corrupt JSON") + } +} + +func TestSaveFeedSyncStateRequiresLinkKey(t *testing.T) { + if err := SaveFeedSyncState(t.TempDir(), &FeedSyncState{}); err == nil { + t.Fatal("SaveFeedSyncState accepted an empty link key") + } + if err := SaveFeedSyncState(t.TempDir(), nil); err == nil { + t.Fatal("SaveFeedSyncState accepted nil state") + } +} + +func TestClearFeedSyncState(t *testing.T) { + dir := t.TempDir() + if err := ClearFeedSyncState(dir); err != nil { + t.Fatalf("ClearFeedSyncState on missing file: %v", err) + } + if err := SaveFeedSyncState(dir, &FeedSyncState{LinkKey: "link"}); err != nil { + t.Fatal(err) + } + if err := ClearFeedSyncState(dir); err != nil { + t.Fatalf("ClearFeedSyncState: %v", err) + } + loaded, err := LoadFeedSyncState(dir) + if err != nil || loaded != nil { + t.Fatalf("after clear: loaded=%+v err=%v, want nil/nil", loaded, err) + } +} From a9ea101106f88bd0163928bd141edb5eb724248e Mon Sep 17 00:00:00 2001 From: Liran Cohen Date: Sun, 26 Jul 2026 15:54:05 +0000 Subject: [PATCH 2/4] fix(engine): converge via the Messages feed with fingerprint-gated anti-entropy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nodes that can speak the Messages interface — the anchor owner (tenant bypass) and wallet delegates holding active protocol-root Messages.Read grants — now run ONE 'feed' subscription stream (MessagesSubscribe, filters [{protocol: mesh}], plural permissionGrantIds, persisted resume cursor) instead of the topology+delivery Records streams. The server's core-protocol shadow filters carry delivery and permission records on the same stream, so it subsumes both. Events remain invalidation-only wake hints; role-only nodes keep the existing two-stream behavior. - RefreshCoordinator: constructor-configurable stream set ([feed] vs [topology, delivery]); operations on unknown streams are ignored; health/mode semantics unchanged. New DegradedReason status field. - SubscriptionWatcher: feed mode with cursor persistence (debounced on events, immediate on EOSE, cleared on 410 ProgressGap) through a feedSyncRecorder that owns the checkpoint; the auth-terminal sentinel (grants revoked/expired mid-stream) drops the stream to uncovered polling fallback and surfaces an operator-facing reason. - DWNControl: periodic-only refresh batches probe the remote feed fingerprint (cids-only MessagesQuery via control.FeedFingerprint) and skip the full 8-phase LoadState when it matches the fingerprint recorded immediately before the last successful load; a full rebuild is still forced at least every 30 minutes. Event-driven, startup, manual, and endpoint reasons always rebuild. - Engine.Config: feed grant IDs + epoch tuples and FeedSyncLoad/Save hooks (the engine never learns file paths); the link key (tenant^endpoint^projectionId^authorizationEpoch) gates cursor resume so grant churn discards stale checkpoints. Refs #267 Co-Authored-By: Claude Fable 5 --- internal/control/dwnclient.go | 22 ++ internal/engine/dwncontrol.go | 96 +++++- internal/engine/dwncontrol_feed_test.go | 184 ++++++++++++ internal/engine/engine.go | 65 +++++ internal/engine/feedsync.go | 250 ++++++++++++++++ internal/engine/feedsync_test.go | 306 ++++++++++++++++++++ internal/engine/refresh_coordinator.go | 95 ++++-- internal/engine/refresh_coordinator_test.go | 71 +++++ internal/engine/subscribe.go | 242 +++++++++++++--- internal/engine/subscribe_test.go | 295 ++++++++++++++++++- 10 files changed, 1546 insertions(+), 80 deletions(-) create mode 100644 internal/engine/dwncontrol_feed_test.go create mode 100644 internal/engine/feedsync.go create mode 100644 internal/engine/feedsync_test.go diff --git a/internal/control/dwnclient.go b/internal/control/dwnclient.go index 360c992..346ad19 100644 --- a/internal/control/dwnclient.go +++ b/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 diff --git a/internal/engine/dwncontrol.go b/internal/engine/dwncontrol.go index 746c34a..edfa110 100644 --- a/internal/engine/dwncontrol.go +++ b/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,7 +335,14 @@ 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 { @@ -311,10 +350,35 @@ func (cc *DWNControl) refreshControlState(ctx context.Context, _ RefreshBatch) e } 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) { diff --git a/internal/engine/dwncontrol_feed_test.go b/internal/engine/dwncontrol_feed_test.go new file mode 100644 index 0000000..40d77e3 --- /dev/null +++ b/internal/engine/dwncontrol_feed_test.go @@ -0,0 +1,184 @@ +package engine + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/enboxorg/meshnet/control/controlclient" + "github.com/enboxorg/meshnet/types/netmap" +) + +type feedGatingHarness struct { + cc *DWNControl + feedSync *feedSyncRecorder + loads *atomic.Int32 + fingerprint *atomic.Pointer[string] + fingerprintErr *atomic.Pointer[error] +} + +func newFeedGatingHarness(t *testing.T) *feedGatingHarness { + t.Helper() + loads := &atomic.Int32{} + fingerprint := &atomic.Pointer[string]{} + initial := "aa11" + fingerprint.Store(&initial) + fingerprintErr := &atomic.Pointer[error]{} + feedSync := newFeedSyncRecorder(testFeedLinkKey, nil, nil) + + cc, err := NewDWNControl( + &DWNControlConfig{ + MapResponseFunc: func(ctx context.Context) (*netmap.NetworkMap, error) { + loads.Add(1) + return &netmap.NetworkMap{}, nil + }, + Streams: []RefreshStream{RefreshStreamFeed}, + FeedFingerprintFunc: func(ctx context.Context) (string, error) { + if errPtr := fingerprintErr.Load(); errPtr != nil { + return "", *errPtr + } + return *fingerprint.Load(), nil + }, + FeedSync: feedSync, + PollInterval: time.Hour, + }, + controlclient.Options{SkipStartForTests: true}, + ) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + t.Cleanup(cc.Shutdown) + return &feedGatingHarness{ + cc: cc, + feedSync: feedSync, + loads: loads, + fingerprint: fingerprint, + fingerprintErr: fingerprintErr, + } +} + +func (h *feedGatingHarness) refresh(t *testing.T, reasons ...RefreshReason) { + t.Helper() + if err := h.cc.refreshControlState(context.Background(), RefreshBatch{Reasons: reasons}); err != nil { + t.Fatalf("refreshControlState(%v): %v", reasons, err) + } +} + +func (h *feedGatingHarness) setFingerprint(fp string) { + h.fingerprint.Store(&fp) +} + +func TestDWNControlFeedFingerprintGatesPeriodicRefreshes(t *testing.T) { + h := newFeedGatingHarness(t) + + // Startup always performs a full load and records the fingerprint + // fetched immediately before it. + h.refresh(t, RefreshReasonStartup) + if got := h.loads.Load(); got != 1 { + t.Fatalf("loads after startup = %d, want 1", got) + } + if got := h.feedSync.storedFingerprint(); got != "aa11" { + t.Fatalf("stored fingerprint = %q, want aa11", got) + } + + // Periodic-only refresh with an unchanged fingerprint skips the load. + h.refresh(t, RefreshReasonPeriodic) + if got := h.loads.Load(); got != 1 { + t.Fatalf("loads after stable periodic = %d, want 1 (skipped)", got) + } + + // A changed fingerprint forces the full load and re-records it. + h.setFingerprint("bb22") + h.refresh(t, RefreshReasonPeriodic) + if got := h.loads.Load(); got != 2 { + t.Fatalf("loads after changed fingerprint = %d, want 2", got) + } + if got := h.feedSync.storedFingerprint(); got != "bb22" { + t.Fatalf("stored fingerprint = %q, want bb22", got) + } +} + +func TestDWNControlFeedNonPeriodicReasonsAlwaysLoad(t *testing.T) { + h := newFeedGatingHarness(t) + h.refresh(t, RefreshReasonStartup) + + // Event-driven, manual, endpoint, and mixed batches never skip, even + // with a stable fingerprint. + for i, reasons := range [][]RefreshReason{ + {RefreshReasonFeed}, + {RefreshReasonManual}, + {RefreshReasonEndpoint}, + {RefreshReasonFeed, RefreshReasonPeriodic}, + } { + h.refresh(t, reasons...) + if got := h.loads.Load(); got != int32(i+2) { + t.Fatalf("loads after %v = %d, want %d", reasons, got, i+2) + } + } +} + +func TestDWNControlFeedForcesFullLoadAfterInterval(t *testing.T) { + h := newFeedGatingHarness(t) + h.refresh(t, RefreshReasonStartup) + + // Stable fingerprint within the interval: skipped. + h.refresh(t, RefreshReasonPeriodic) + if got := h.loads.Load(); got != 1 { + t.Fatalf("loads = %d, want 1", got) + } + + // Belt and braces: past the forced-rebuild interval a stable + // fingerprint no longer skips. + h.cc.lastFullRefreshAt = time.Now().Add(-h.cc.config.FeedFullRefreshInterval - time.Minute) + h.refresh(t, RefreshReasonPeriodic) + if got := h.loads.Load(); got != 2 { + t.Fatalf("loads after forced interval = %d, want 2", got) + } + + // The forced load refreshes the timestamp, re-arming the skip. + h.refresh(t, RefreshReasonPeriodic) + if got := h.loads.Load(); got != 2 { + t.Fatalf("loads after re-armed skip = %d, want 2", got) + } +} + +func TestDWNControlFeedFingerprintProbeFailureDisablesSkip(t *testing.T) { + h := newFeedGatingHarness(t) + h.refresh(t, RefreshReasonStartup) + + // A failing probe cannot prove stability: the load proceeds and the + // stored fingerprint is cleared (unknown). + probeErr := errors.New("probe failed") + h.fingerprintErr.Store(&probeErr) + h.refresh(t, RefreshReasonPeriodic) + if got := h.loads.Load(); got != 2 { + t.Fatalf("loads after probe failure = %d, want 2", got) + } + if got := h.feedSync.storedFingerprint(); got != "" { + t.Fatalf("stored fingerprint = %q, want cleared", got) + } + + // Probe recovers with the original fingerprint: the previous entry was + // cleared, so one more full load re-establishes the baseline. + h.fingerprintErr.Store(nil) + h.refresh(t, RefreshReasonPeriodic) + if got := h.loads.Load(); got != 3 { + t.Fatalf("loads after probe recovery = %d, want 3", got) + } + h.refresh(t, RefreshReasonPeriodic) + if got := h.loads.Load(); got != 3 { + t.Fatalf("loads after restored baseline = %d, want 3 (skipped)", got) + } +} + +func TestDWNControlFeedEmptyServerFingerprintNeverSkips(t *testing.T) { + h := newFeedGatingHarness(t) + h.setFingerprint("") + h.refresh(t, RefreshReasonStartup) + h.refresh(t, RefreshReasonPeriodic) + if got := h.loads.Load(); got != 2 { + t.Fatalf("loads with empty fingerprints = %d, want 2 (no skipping)", got) + } +} diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 7885381..9841fd9 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -20,6 +20,7 @@ import ( "github.com/enboxorg/meshd/internal/dwn" dwncrypto "github.com/enboxorg/meshd/internal/dwn/crypto" "github.com/enboxorg/meshd/internal/mesh" + "github.com/enboxorg/meshd/internal/state" "github.com/tailscale/wireguard-go/tun" @@ -74,6 +75,10 @@ type Engine struct { snapshots *meshSnapshotStore refreshCoordinator atomic.Pointer[RefreshCoordinator] + + // feedSync owns the persisted replication-feed checkpoint in feed mode. + // nil in Records-stream mode. + feedSync *feedSyncRecorder } // Config holds the configuration for creating an Engine. @@ -155,6 +160,26 @@ type Config struct { // WritePermissionGrantID. WriteDelegatedGrant json.RawMessage + // FeedGrantIDs are the active protocol-root Messages.Read grant IDs the + // engine invokes for replication-feed reads, in canonical (sorted, + // deduplicated) order. Non-empty enables feed mode for wallet delegates; + // the anchor owner (Signer DID == AnchorTenant) needs none. Empty on a + // non-owner keeps the two Records streams. + FeedGrantIDs []string + + // FeedGrants are the grant tuples hashed into the feed link's + // authorization epoch. Must describe the same grants as FeedGrantIDs. + FeedGrants []dwn.SyncAuthorizationGrant + + // FeedSyncLoad loads the persisted feed replication checkpoint (nil when + // absent). The engine never learns the storage location; callers close + // over it. Nil disables resume across restarts. + FeedSyncLoad func() (*state.FeedSyncState, error) + + // FeedSyncSave persists the feed replication checkpoint. Nil disables + // persistence (feed mode still works, without cross-restart resume). + FeedSyncSave func(state.FeedSyncState) error + // GrantKeys holds the delegate's grant-key subtree decrypters for // decrypting protocol records (enbox connect sessions). GrantKeys *control.GrantKeySet @@ -306,6 +331,31 @@ func New(cfg Config) (*Engine, error) { controlOpts..., ) + // Feed mode: owner/anchor nodes and delegates holding protocol-root + // Messages.Read grants converge via the single Messages replication-feed + // stream. The persisted checkpoint resumes only under an identical link + // key — grant churn or expiry changes the authorization epoch and + // discards the stale cursor. + feedMode := feedModeEnabled(cfg) + var feedRecorder *feedSyncRecorder + if feedMode { + linkKey, err := feedLinkKey(cfg) + if err != nil { + return nil, fmt.Errorf("deriving feed link key: %w", err) + } + feedRecorder = newFeedSyncRecorder(linkKey, cfg.FeedSyncSave, l) + if cfg.FeedSyncLoad != nil { + loaded, loadErr := cfg.FeedSyncLoad() + if loadErr != nil { + l.Warn("loading feed sync state failed; starting without a resume cursor", + slog.Any("error", loadErr), + ) + } else { + feedRecorder.adopt(loaded) + } + } + } + // Create the converter that bridges meshd types to meshnet types. converter := NewConverter(domain, WithConverterLogger(l)) converter.MagicDNSSuffix = magicDNS @@ -470,8 +520,11 @@ func New(cfg Config) (*Engine, error) { SelfDID: cfg.SelfDID, Signer: cfg.Signer, ReadAuth: readAuth, + FeedMode: feedMode, + FeedGrantIDs: cfg.FeedGrantIDs, Logger: l, }) + subWatcher.feedSync = feedRecorder // Wire the DWN control client into the LocalBackend. // MapResponseFunc closes over our DWNClient and Converter to produce @@ -500,6 +553,14 @@ func New(cfg Config) (*Engine, error) { } }, } + if feedMode { + feedGrantIDs := append([]string(nil), cfg.FeedGrantIDs...) + dwnControlConfig.Streams = []RefreshStream{RefreshStreamFeed} + dwnControlConfig.FeedFingerprintFunc = func(ctx context.Context) (string, error) { + return dwnClient.FeedFingerprint(ctx, feedGrantIDs) + } + dwnControlConfig.FeedSync = feedRecorder + } // If the caller provided a WireGuard private key (already published to // DWN), inject it so the engine uses the same key. Without this, meshnet @@ -535,6 +596,7 @@ func New(cfg Config) (*Engine, error) { osRouter: osRouter, logger: l, snapshots: snapshots, + feedSync: feedRecorder, } engineRef.initializeRoutingStatus(cfg.TUNName != "") return engineRef, nil @@ -617,6 +679,9 @@ func (e *Engine) Stop() error { e.subWatcher.Stop() } + // Flush the feed checkpoint after the watcher stops delivering cursors. + e.feedSync.close() + if e.cancel != nil { e.cancel() } diff --git a/internal/engine/feedsync.go b/internal/engine/feedsync.go new file mode 100644 index 0000000..6658894 --- /dev/null +++ b/internal/engine/feedsync.go @@ -0,0 +1,250 @@ +package engine + +import ( + "log/slog" + "sync" + "time" + + "github.com/enboxorg/meshd/internal/dwn" + "github.com/enboxorg/meshd/internal/state" + "github.com/enboxorg/meshd/protocols" +) + +// feedCursorPersistDebounce coalesces cursor persistence during event bursts. +// EOSE, progress gaps, and fingerprint updates persist immediately. +const feedCursorPersistDebounce = 2 * time.Second + +// feedSyncRecorder owns the in-memory replication-feed checkpoint (resume +// cursor + last-verified fingerprint) and serializes its persistence through +// the caller-supplied save hook. The subscription watcher records cursor +// advances; the control client records fingerprints. Neither learns where the +// checkpoint is stored. +type feedSyncRecorder struct { + linkKey string + save func(state.FeedSyncState) error + debounce time.Duration + logger *slog.Logger + + mu sync.Mutex + cursor *dwn.ProgressToken + fingerprint string + dirty bool + timer *time.Timer + closed bool +} + +func newFeedSyncRecorder(linkKey string, save func(state.FeedSyncState) error, logger *slog.Logger) *feedSyncRecorder { + if logger == nil { + logger = slog.Default() + } + return &feedSyncRecorder{ + linkKey: linkKey, + save: save, + debounce: feedCursorPersistDebounce, + logger: logger.With(slog.String("component", "feed-sync")), + } +} + +// adopt restores a persisted checkpoint. Checkpoints from a different link +// key (grant churn, endpoint change, scope change) are discarded: their +// cursor belongs to a stream the current authorization epoch cannot resume. +func (r *feedSyncRecorder) adopt(loaded *state.FeedSyncState) { + if r == nil || loaded == nil || loaded.LinkKey != r.linkKey { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if loaded.Cursor.Valid() { + r.cursor = cloneProgressToken(loaded.Cursor) + } + r.fingerprint = loaded.Fingerprint +} + +// resumeCursor returns the cursor the initial feed subscribe should resume +// from, or nil to replay from scratch. +func (r *feedSyncRecorder) resumeCursor() *dwn.ProgressToken { + if r == nil { + return nil + } + r.mu.Lock() + defer r.mu.Unlock() + return cloneProgressToken(r.cursor) +} + +// recordCursor notes an event-delivered cursor advance and schedules a +// debounced persist. +func (r *feedSyncRecorder) recordCursor(cursor *dwn.ProgressToken) { + if r == nil || !cursor.Valid() { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if r.closed { + return + } + r.cursor = cloneProgressToken(cursor) + r.dirty = true + if r.timer == nil { + r.timer = time.AfterFunc(r.debounce, r.persistDebounced) + } +} + +// flushCursor persists a cursor immediately (EOSE barrier). +func (r *feedSyncRecorder) flushCursor(cursor *dwn.ProgressToken) { + if r == nil || !cursor.Valid() { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if r.closed { + return + } + r.cursor = cloneProgressToken(cursor) + r.dirty = true + r.stopTimerLocked() + r.persistLocked() +} + +// clearCursor discards the persisted cursor after a 410 progress gap so the +// reconnect replays from scratch. The fingerprint is retained: it describes +// feed contents, not log positions. +func (r *feedSyncRecorder) clearCursor() { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if r.closed { + return + } + r.cursor = nil + r.dirty = true + r.stopTimerLocked() + r.persistLocked() +} + +// storedFingerprint returns the fingerprint recorded before the last +// successful full state load, or "" when unknown. +func (r *feedSyncRecorder) storedFingerprint() string { + if r == nil { + return "" + } + r.mu.Lock() + defer r.mu.Unlock() + return r.fingerprint +} + +// recordFingerprint persists the fingerprint observed immediately before a +// successful full state load. An empty value marks the fingerprint unknown +// and disables periodic skipping until the next full load. +func (r *feedSyncRecorder) recordFingerprint(fingerprint string) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if r.closed { + return + } + r.fingerprint = fingerprint + r.dirty = true + r.persistLocked() +} + +// close flushes any pending cursor advance and stops further persistence. +func (r *feedSyncRecorder) close() { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + if r.closed { + return + } + r.closed = true + r.stopTimerLocked() + if r.dirty { + r.persistLocked() + } +} + +func (r *feedSyncRecorder) persistDebounced() { + r.mu.Lock() + defer r.mu.Unlock() + r.timer = nil + if r.closed || !r.dirty { + return + } + r.persistLocked() +} + +func (r *feedSyncRecorder) stopTimerLocked() { + if r.timer != nil { + r.timer.Stop() + r.timer = nil + } +} + +func (r *feedSyncRecorder) persistLocked() { + if r.save == nil { + r.dirty = false + return + } + err := r.save(state.FeedSyncState{ + LinkKey: r.linkKey, + Cursor: cloneProgressToken(r.cursor), + Fingerprint: r.fingerprint, + }) + if err != nil { + r.logger.Warn("persisting feed sync state failed", slog.Any("error", err)) + return + } + r.dirty = false +} + +func cloneProgressToken(token *dwn.ProgressToken) *dwn.ProgressToken { + if token == nil { + return nil + } + clone := *token + return &clone +} + +// feedModeEnabled reports whether the engine converges via the Messages +// replication feed: the anchor owner signs Messages operations tenant-side +// (grant bypass), and wallet delegates need active protocol-root +// Messages.Read grants. Everyone else keeps the two Records streams. +func feedModeEnabled(cfg Config) bool { + if cfg.Signer == nil { + return false + } + if cfg.Signer.DID != "" && cfg.Signer.DID == cfg.AnchorTenant { + return true + } + return len(cfg.FeedGrantIDs) > 0 +} + +// feedLinkKey derives the replication link identity of the engine's feed +// subscription: (tenant, endpoint, projectionId, authorizationEpoch). Grant +// churn or expiry changes the authorization epoch, which invalidates any +// persisted cursor recorded under the previous link key. +func feedLinkKey(cfg Config) (string, error) { + scope, err := dwn.ProtocolSetSyncScope([]string{protocols.MeshProtocolURI}) + if err != nil { + return "", err + } + projectionID, err := dwn.ComputeProjectionID(cfg.AnchorTenant, scope) + if err != nil { + return "", err + } + var epoch string + if cfg.Signer.DID == cfg.AnchorTenant { + epoch = dwn.ComputeOwnerAuthorizationEpoch() + } else { + epoch, err = dwn.ComputeDelegateAuthorizationEpoch(cfg.Signer.DID, cfg.FeedGrants) + if err != nil { + return "", err + } + } + return dwn.BuildLinkKey(cfg.AnchorTenant, cfg.AnchorEndpoint, projectionID, epoch), nil +} diff --git a/internal/engine/feedsync_test.go b/internal/engine/feedsync_test.go new file mode 100644 index 0000000..dbecc60 --- /dev/null +++ b/internal/engine/feedsync_test.go @@ -0,0 +1,306 @@ +package engine + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/enboxorg/meshd/internal/dwn" + "github.com/enboxorg/meshd/internal/state" +) + +const testFeedLinkKey = "did:dht:owner^https://dwn.example^proj^epoch" + +type feedSaveRecorder struct { + mu sync.Mutex + saves []state.FeedSyncState + err error +} + +func (r *feedSaveRecorder) save(fs state.FeedSyncState) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.err != nil { + return r.err + } + r.saves = append(r.saves, fs) + return nil +} + +func (r *feedSaveRecorder) takeSaves() []state.FeedSyncState { + r.mu.Lock() + defer r.mu.Unlock() + saves := append([]state.FeedSyncState(nil), r.saves...) + r.saves = nil + return saves +} + +func testCursor(position string) *dwn.ProgressToken { + return &dwn.ProgressToken{StreamID: "aabbccddeeff0011", Epoch: "epoch-1", Position: position} +} + +func TestFeedSyncRecorderAdoptRequiresMatchingLinkKey(t *testing.T) { + tests := []struct { + name string + loaded *state.FeedSyncState + want *dwn.ProgressToken + }{ + {name: "nil state", loaded: nil, want: nil}, + { + name: "matching link key adopts cursor", + loaded: &state.FeedSyncState{LinkKey: testFeedLinkKey, Cursor: testCursor("7"), Fingerprint: "aa"}, + want: testCursor("7"), + }, + { + name: "link key mismatch discards cursor", + loaded: &state.FeedSyncState{LinkKey: "other^link^key^x", Cursor: testCursor("7")}, + want: nil, + }, + { + name: "invalid cursor discarded", + loaded: &state.FeedSyncState{LinkKey: testFeedLinkKey, Cursor: &dwn.ProgressToken{Position: "7"}}, + want: nil, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + r := newFeedSyncRecorder(testFeedLinkKey, nil, nil) + r.adopt(test.loaded) + got := r.resumeCursor() + switch { + case test.want == nil && got != nil: + t.Fatalf("resume cursor = %+v, want nil", got) + case test.want != nil && (got == nil || *got != *test.want): + t.Fatalf("resume cursor = %+v, want %+v", got, test.want) + } + }) + } + + t.Run("matching link key adopts fingerprint", func(t *testing.T) { + r := newFeedSyncRecorder(testFeedLinkKey, nil, nil) + r.adopt(&state.FeedSyncState{LinkKey: testFeedLinkKey, Fingerprint: "ff00"}) + if got := r.storedFingerprint(); got != "ff00" { + t.Fatalf("stored fingerprint = %q, want ff00", got) + } + }) + t.Run("mismatched link key discards fingerprint", func(t *testing.T) { + r := newFeedSyncRecorder(testFeedLinkKey, nil, nil) + r.adopt(&state.FeedSyncState{LinkKey: "different", Fingerprint: "ff00"}) + if got := r.storedFingerprint(); got != "" { + t.Fatalf("stored fingerprint = %q, want empty", got) + } + }) +} + +func TestFeedSyncRecorderDebouncesCursorPersists(t *testing.T) { + saves := &feedSaveRecorder{} + r := newFeedSyncRecorder(testFeedLinkKey, saves.save, nil) + r.debounce = 20 * time.Millisecond + + r.recordCursor(testCursor("1")) + r.recordCursor(testCursor("2")) + r.recordCursor(testCursor("3")) + if got := saves.takeSaves(); len(got) != 0 { + t.Fatalf("saves before debounce = %d, want 0", len(got)) + } + + deadline := time.Now().Add(2 * time.Second) + var persisted []state.FeedSyncState + for len(persisted) == 0 { + if time.Now().After(deadline) { + t.Fatal("debounced persist never ran") + } + time.Sleep(5 * time.Millisecond) + persisted = saves.takeSaves() + } + if len(persisted) != 1 { + t.Fatalf("saves = %d, want one coalesced persist", len(persisted)) + } + got := persisted[0] + if got.LinkKey != testFeedLinkKey || got.Cursor == nil || got.Cursor.Position != "3" { + t.Fatalf("persisted = %+v, want latest cursor position 3", got) + } +} + +func TestFeedSyncRecorderFlushAndClearPersistImmediately(t *testing.T) { + saves := &feedSaveRecorder{} + r := newFeedSyncRecorder(testFeedLinkKey, saves.save, nil) + r.debounce = time.Hour // debounced persists must not fire in this test + + r.recordCursor(testCursor("1")) // pending, not yet persisted + r.flushCursor(testCursor("2")) + flushed := saves.takeSaves() + if len(flushed) != 1 || flushed[0].Cursor == nil || flushed[0].Cursor.Position != "2" { + t.Fatalf("flush saves = %+v, want one save at position 2", flushed) + } + + r.recordFingerprint("ff00") + withFingerprint := saves.takeSaves() + if len(withFingerprint) != 1 || withFingerprint[0].Fingerprint != "ff00" { + t.Fatalf("fingerprint saves = %+v, want one save with ff00", withFingerprint) + } + if withFingerprint[0].Cursor == nil || withFingerprint[0].Cursor.Position != "2" { + t.Fatalf("fingerprint save cursor = %+v, want retained position 2", withFingerprint[0].Cursor) + } + + // A progress gap clears the cursor but keeps the fingerprint: it + // describes feed contents, not log positions. + r.clearCursor() + cleared := saves.takeSaves() + if len(cleared) != 1 || cleared[0].Cursor != nil || cleared[0].Fingerprint != "ff00" { + t.Fatalf("clear saves = %+v, want nil cursor and retained fingerprint", cleared) + } + if r.resumeCursor() != nil { + t.Fatal("resume cursor survived clearCursor") + } +} + +func TestFeedSyncRecorderCloseFlushesPendingCursor(t *testing.T) { + saves := &feedSaveRecorder{} + r := newFeedSyncRecorder(testFeedLinkKey, saves.save, nil) + r.debounce = time.Hour + + r.recordCursor(testCursor("9")) + r.close() + got := saves.takeSaves() + if len(got) != 1 || got[0].Cursor == nil || got[0].Cursor.Position != "9" { + t.Fatalf("close saves = %+v, want pending cursor flushed", got) + } + + // Closed recorders ignore further mutations. + r.recordCursor(testCursor("10")) + r.recordFingerprint("aa") + if got := saves.takeSaves(); len(got) != 0 { + t.Fatalf("saves after close = %+v, want none", got) + } +} + +func TestFeedSyncRecorderSurvivesSaveErrors(t *testing.T) { + saves := &feedSaveRecorder{err: errors.New("disk full")} + r := newFeedSyncRecorder(testFeedLinkKey, saves.save, nil) + r.flushCursor(testCursor("1")) + r.recordFingerprint("aa") + + saves.mu.Lock() + saves.err = nil + saves.mu.Unlock() + r.recordFingerprint("bb") + got := saves.takeSaves() + if len(got) != 1 || got[0].Fingerprint != "bb" || got[0].Cursor == nil { + t.Fatalf("saves after recovery = %+v, want cursor+fingerprint", got) + } +} + +func TestFeedModeEnabled(t *testing.T) { + grants := []dwn.SyncAuthorizationGrant{{ + ID: "grant-feed", + DateGranted: "2026-07-01T00:00:00.000000Z", + DateExpires: "2026-08-01T00:00:00.000000Z", + }} + tests := []struct { + name string + cfg Config + want bool + }{ + { + name: "owner signer equals anchor tenant", + cfg: Config{ + AnchorTenant: "did:dht:owner", + SelfDID: "did:dht:owner", + Signer: &dwn.Signer{DID: "did:dht:owner"}, + }, + want: true, + }, + { + name: "delegate with feed grants", + cfg: Config{ + AnchorTenant: "did:dht:owner", + SelfDID: "did:jwk:node", + Signer: &dwn.Signer{DID: "did:jwk:delegate"}, + FeedGrantIDs: []string{"grant-feed"}, + FeedGrants: grants, + }, + want: true, + }, + { + name: "role-only node keeps records streams", + cfg: Config{ + AnchorTenant: "did:dht:owner", + SelfDID: "did:jwk:node", + Signer: &dwn.Signer{DID: "did:jwk:node"}, + }, + want: false, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if got := feedModeEnabled(test.cfg); got != test.want { + t.Fatalf("feedModeEnabled = %v, want %v", got, test.want) + } + }) + } +} + +func TestFeedLinkKey(t *testing.T) { + ownerCfg := Config{ + AnchorEndpoint: "https://dwn.example", + AnchorTenant: "did:dht:owner", + Signer: &dwn.Signer{DID: "did:dht:owner"}, + } + ownerKey, err := feedLinkKey(ownerCfg) + if err != nil { + t.Fatalf("owner feedLinkKey: %v", err) + } + + grants := []dwn.SyncAuthorizationGrant{{ + ID: "grant-feed", + DateGranted: "2026-07-01T00:00:00.000000Z", + DateExpires: "2026-08-01T00:00:00.000000Z", + }} + delegateCfg := Config{ + AnchorEndpoint: "https://dwn.example", + AnchorTenant: "did:dht:owner", + Signer: &dwn.Signer{DID: "did:jwk:delegate"}, + FeedGrantIDs: []string{"grant-feed"}, + FeedGrants: grants, + } + delegateKey, err := feedLinkKey(delegateCfg) + if err != nil { + t.Fatalf("delegate feedLinkKey: %v", err) + } + if ownerKey == delegateKey { + t.Fatal("owner and delegate link keys must differ (different authorization epochs)") + } + + // Deterministic: same inputs, same key. + again, err := feedLinkKey(delegateCfg) + if err != nil { + t.Fatal(err) + } + if again != delegateKey { + t.Fatalf("link key not deterministic: %q vs %q", again, delegateKey) + } + + // Grant churn (different expiry) changes the epoch and the link key. + churned := delegateCfg + churned.FeedGrants = []dwn.SyncAuthorizationGrant{{ + ID: "grant-feed", + DateGranted: "2026-07-01T00:00:00.000000Z", + DateExpires: "2026-09-01T00:00:00.000000Z", + }} + churnedKey, err := feedLinkKey(churned) + if err != nil { + t.Fatal(err) + } + if churnedKey == delegateKey { + t.Fatal("grant churn must change the link key") + } + + // Delegate mode without grant tuples is a wiring error. + broken := delegateCfg + broken.FeedGrants = nil + if _, err := feedLinkKey(broken); err == nil { + t.Fatal("feedLinkKey accepted delegate config without grant tuples") + } +} diff --git a/internal/engine/refresh_coordinator.go b/internal/engine/refresh_coordinator.go index daa7e49..874f98d 100644 --- a/internal/engine/refresh_coordinator.go +++ b/internal/engine/refresh_coordinator.go @@ -21,6 +21,7 @@ const ( RefreshReasonPeriodic RefreshReason = "periodic" RefreshReasonTopology RefreshReason = "topology" RefreshReasonDelivery RefreshReason = "delivery" + RefreshReasonFeed RefreshReason = "feed" RefreshReasonEndpoint RefreshReason = "endpoint" RefreshReasonManual RefreshReason = "manual" ) @@ -31,6 +32,12 @@ type RefreshStream string const ( RefreshStreamTopology RefreshStream = "topology" RefreshStreamDelivery RefreshStream = "delivery" + + // RefreshStreamFeed is the single Messages replication-log stream used by + // nodes eligible for feed mode. Its core-protocol shadow filters carry the + // delivery and permission records too, so it subsumes both Records + // streams. + RefreshStreamFeed RefreshStream = "feed" ) // RefreshBatch is the immutable set of invalidations represented by one @@ -63,12 +70,17 @@ type RefreshClock interface { type RefreshCoordinatorConfig struct { Refresh RefreshFunc - // FallbackInterval is used whenever either required stream is uncovered, + // Streams is the authoritative subscription stream set this coordinator + // tracks. Stream operations naming a stream outside this set are ignored. + // Nil defaults to the two Records streams (topology, delivery). + Streams []RefreshStream + + // FallbackInterval is used whenever any required stream is uncovered, // disconnected, or has not yet been repaired. Default: 30 seconds. FallbackInterval time.Duration - // HealthyInterval is the anti-entropy interval when both topology and - // delivery streams are covered, live, and repaired. Default: 5 minutes. + // HealthyInterval is the anti-entropy interval when every required stream + // is covered, live, and repaired. Default: 5 minutes. HealthyInterval time.Duration // Debounce waits for a quiet window after event invalidations. MaxDebounce @@ -112,11 +124,15 @@ type RefreshStreamHealth struct { // RefreshCoordinatorHealth is a point-in-time, deep-copied scheduler view. type RefreshCoordinatorHealth struct { - Running bool - Paused bool - InFlight bool - Mode RefreshMode - StreamsHealthy bool + Running bool + Paused bool + InFlight bool + Mode RefreshMode + StreamsHealthy bool + // DegradedReason is an operator-facing explanation of why subscription + // coverage was lost terminally (e.g. wallet session grants expired). + // Empty while coverage is intact. + DegradedReason string Streams map[RefreshStream]RefreshStreamHealth PendingReasons []RefreshReason ConsecutiveFailures int @@ -156,10 +172,11 @@ type RefreshCoordinator struct { retryJitter func(time.Duration) time.Duration clock RefreshClock - mu sync.Mutex - pending map[RefreshReason]refreshPending - streams map[RefreshStream]refreshStreamState - wake chan struct{} + mu sync.Mutex + pending map[RefreshReason]refreshPending + streams map[RefreshStream]refreshStreamState + degradedReason string + wake chan struct{} sequence uint64 started bool @@ -231,6 +248,18 @@ func NewRefreshCoordinator(cfg RefreshCoordinatorConfig) (*RefreshCoordinator, e retryJitter = defaultRefreshRetryJitter } + streamSet := cfg.Streams + if len(streamSet) == 0 { + streamSet = []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} + } + streams := make(map[RefreshStream]refreshStreamState, len(streamSet)) + for _, stream := range streamSet { + if stream == "" { + return nil, fmt.Errorf("refresh coordinator stream names must not be empty") + } + streams[stream] = refreshStreamState{} + } + return &RefreshCoordinator{ refresh: cfg.Refresh, fallbackInterval: fallbackInterval, @@ -243,11 +272,8 @@ func NewRefreshCoordinator(cfg RefreshCoordinatorConfig) (*RefreshCoordinator, e retryJitter: retryJitter, clock: clock, pending: make(map[RefreshReason]refreshPending), - streams: map[RefreshStream]refreshStreamState{ - RefreshStreamTopology: {}, - RefreshStreamDelivery: {}, - }, - wake: make(chan struct{}, 1), + streams: streams, + wake: make(chan struct{}, 1), }, nil } @@ -354,7 +380,11 @@ func (c *RefreshCoordinator) SetStreamCovered(stream RefreshStream, covered bool c.mu.Unlock() return } - state := c.streams[stream] + state, known := c.streams[stream] + if !known { + c.mu.Unlock() + return + } if state.covered == covered { c.mu.Unlock() return @@ -393,7 +423,11 @@ func (c *RefreshCoordinator) SetStreamLive(stream RefreshStream, live, needsFull c.mu.Unlock() return } - state := c.streams[stream] + state, known := c.streams[stream] + if !known { + c.mu.Unlock() + return + } wasLive := state.live changed := wasLive != live state.live = live @@ -440,7 +474,11 @@ func (c *RefreshCoordinator) InvalidateStream(stream RefreshStream, reason Refre c.mu.Unlock() return } - state := c.streams[stream] + state, known := c.streams[stream] + if !known { + c.mu.Unlock() + return + } c.sequence++ state.repaired = false state.repairSequence = c.sequence @@ -454,6 +492,17 @@ func (c *RefreshCoordinator) InvalidateStream(stream RefreshStream, reason Refre c.signal() } +// SetDegradedReason records (or clears, with an empty string) the +// operator-facing explanation surfaced through Health and the daemon status +// when subscription coverage is lost terminally. +func (c *RefreshCoordinator) SetDegradedReason(reason string) { + c.mu.Lock() + if !c.stopped { + c.degradedReason = reason + } + c.mu.Unlock() +} + // Health returns a deep copy. Mutating its maps or slices cannot affect the // coordinator. func (c *RefreshCoordinator) Health() RefreshCoordinatorHealth { @@ -467,6 +516,7 @@ func (c *RefreshCoordinator) Health() RefreshCoordinatorHealth { InFlight: c.inFlight, Mode: modeForHealth(healthy), StreamsHealthy: healthy, + DegradedReason: c.degradedReason, Streams: make(map[RefreshStream]RefreshStreamHealth, len(c.streams)), PendingReasons: sortedPendingReasons(c.pending), ConsecutiveFailures: c.consecutiveFailures, @@ -705,8 +755,7 @@ func (c *RefreshCoordinator) currentIntervalLocked() time.Duration { } func (c *RefreshCoordinator) streamsHealthyLocked() bool { - for _, source := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} { - state := c.streams[source] + for _, state := range c.streams { if !state.covered || !state.live || !state.repaired { return false } @@ -727,6 +776,8 @@ func reasonForStream(stream RefreshStream) RefreshReason { return RefreshReasonTopology case RefreshStreamDelivery: return RefreshReasonDelivery + case RefreshStreamFeed: + return RefreshReasonFeed default: return RefreshReason(stream) } diff --git a/internal/engine/refresh_coordinator_test.go b/internal/engine/refresh_coordinator_test.go index f3412e0..17fbba0 100644 --- a/internal/engine/refresh_coordinator_test.go +++ b/internal/engine/refresh_coordinator_test.go @@ -530,6 +530,77 @@ func TestRefreshCoordinatorExponentialBackoffCaps(t *testing.T) { } } +func TestRefreshCoordinatorConfigurableStreamSet(t *testing.T) { + clock := newCoordinatorFakeClock() + started := make(chan RefreshBatch, 4) + coordinator, err := NewRefreshCoordinator(RefreshCoordinatorConfig{ + Refresh: func(_ context.Context, batch RefreshBatch) error { + started <- batch + return nil + }, + Streams: []RefreshStream{RefreshStreamFeed}, + FallbackInterval: 10 * time.Second, + HealthyInterval: 100 * time.Second, + Jitter: identityRefreshJitter, + RetryJitter: identityRefreshJitter, + Clock: clock, + }) + if err != nil { + t.Fatal(err) + } + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + receiveBatch(t, started) + waitCoordinator(t, func() bool { return !coordinator.Health().InFlight }) + + health := coordinator.Health() + if len(health.Streams) != 1 { + t.Fatalf("streams = %v, want only feed", health.Streams) + } + if _, ok := health.Streams[RefreshStreamFeed]; !ok { + t.Fatalf("streams = %v, missing feed", health.Streams) + } + if health.Mode != RefreshModeFallback { + t.Fatalf("mode = %v, want fallback before coverage", health.Mode) + } + + // Operations on streams outside the configured set are ignored and can + // never affect health or scheduling. + coordinator.SetStreamCovered(RefreshStreamTopology, true) + coordinator.SetStreamLive(RefreshStreamTopology, true, true) + coordinator.InvalidateStream(RefreshStreamTopology, RefreshReasonTopology) + assertNoBatch(t, started) + if got := coordinator.Health().Streams; len(got) != 1 { + t.Fatalf("streams after unknown-stream ops = %v, want only feed", got) + } + + // The single feed stream alone drives healthy mode. + coordinator.SetStreamCovered(RefreshStreamFeed, true) + coordinator.SetStreamLive(RefreshStreamFeed, true, true) + batch := receiveBatch(t, started) + assertReasons(t, batch, RefreshReasonFeed) + waitCoordinator(t, func() bool { return coordinator.Health().StreamsHealthy }) + if got := coordinator.Health().Mode; got != RefreshModeHealthy { + t.Fatalf("mode = %v, want healthy with the feed stream repaired", got) + } +} + +func TestRefreshCoordinatorDegradedReason(t *testing.T) { + clock := newCoordinatorFakeClock() + coordinator := newTestRefreshCoordinator(t, clock, func(context.Context, RefreshBatch) error { return nil }) + + coordinator.SetDegradedReason("grants expired") + if got := coordinator.Health().DegradedReason; got != "grants expired" { + t.Fatalf("degraded reason = %q, want set", got) + } + coordinator.SetDegradedReason("") + if got := coordinator.Health().DegradedReason; got != "" { + t.Fatalf("degraded reason = %q, want cleared", got) + } +} + func newTestRefreshCoordinator(t *testing.T, clock RefreshClock, refresh RefreshFunc) *RefreshCoordinator { t.Helper() coordinator, err := NewRefreshCoordinator(RefreshCoordinatorConfig{ diff --git a/internal/engine/subscribe.go b/internal/engine/subscribe.go index ea46005..9449933 100644 --- a/internal/engine/subscribe.go +++ b/internal/engine/subscribe.go @@ -11,6 +11,7 @@ package engine import ( "context" + "errors" "fmt" "log/slog" "sync" @@ -30,6 +31,14 @@ type subscriptionManager interface { dwn.EventHandler, dwn.SubscriptionLifecycleHandler, ) (*dwn.Subscription, error) + SubscribeMessagesWithLifecycle( + context.Context, + string, + *dwn.Signer, + dwn.MessagesSubscribeOptions, + dwn.EventHandler, + dwn.SubscriptionLifecycleHandler, + ) (*dwn.Subscription, error) CloseAll() } @@ -45,8 +54,13 @@ type subscriptionRefreshCoordinator interface { SetStreamLive(RefreshStream, bool, bool) InvalidateStream(RefreshStream, RefreshReason) Notify(RefreshReason) + SetDegradedReason(string) } +// feedAuthDegradedReason is surfaced through the daemon status when the feed +// subscription dies terminally with the delivery-authorization sentinel. +const feedAuthDegradedReason = "wallet session grants expired or revoked — run 'meshd auth connect' to refresh them" + // SubscriptionWatcher subscribes to DWN record changes and reports stream // health and invalidations to the engine's refresh coordinator. // @@ -63,6 +77,17 @@ type SubscriptionWatcher struct { logger *slog.Logger newManager subscriptionManagerFactory + // feedMode replaces the two Records streams with the single Messages + // replication-feed stream. feedSync (optional) persists the resume + // cursor across restarts. + feedMode bool + feedGrantIDs []string + feedSync *feedSyncRecorder + + // streamNames is the fixed stream set this watcher reports on: + // [feed] in feed mode, [topology, delivery] otherwise. + streamNames []RefreshStream + mu sync.Mutex manager subscriptionManager cancel context.CancelFunc @@ -100,6 +125,17 @@ type SubscriptionWatcherConfig struct { // subscriptions and therefore keeps polling for topology changes. ReadAuth dwn.MessageAuth + // FeedMode runs the single Messages replication-feed stream instead of + // the two Records streams. The server's core-protocol shadow filters make + // that one stream carry $encryption/delivery and mesh-tagged permission + // records too. Requires the anchor owner signing identity or active + // protocol-root Messages.Read grants. + FeedMode bool + + // FeedGrantIDs is the canonical permissionGrantIds list invoked on the + // feed subscription. Empty for the anchor owner (tenant bypass). + FeedGrantIDs []string + // Logger is the structured logger. Logger *slog.Logger } @@ -112,6 +148,15 @@ func NewSubscriptionWatcher(cfg SubscriptionWatcherConfig) *SubscriptionWatcher l = slog.Default() } + streamNames := []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} + if cfg.FeedMode { + streamNames = []RefreshStream{RefreshStreamFeed} + } + streams := make(map[RefreshStream]subscriptionWatcherStreamState, len(streamNames)) + for _, stream := range streamNames { + streams[stream] = subscriptionWatcherStreamState{} + } + return &SubscriptionWatcher{ endpoint: cfg.AnchorEndpoint, anchorTenant: cfg.AnchorTenant, @@ -119,14 +164,14 @@ func NewSubscriptionWatcher(cfg SubscriptionWatcherConfig) *SubscriptionWatcher selfDID: cfg.SelfDID, signer: cfg.Signer, readAuth: cloneMessageAuth(cfg.ReadAuth), + feedMode: cfg.FeedMode, + feedGrantIDs: append([]string(nil), cfg.FeedGrantIDs...), + streamNames: streamNames, logger: l.With(slog.String("component", "subscription-watcher")), newManager: func(endpoint string, logger *slog.Logger) subscriptionManager { return dwn.NewSubscriptionManager(endpoint, logger) }, - streams: map[RefreshStream]subscriptionWatcherStreamState{ - RefreshStreamTopology: {}, - RefreshStreamDelivery: {}, - }, + streams: streams, } } @@ -154,10 +199,10 @@ func (w *SubscriptionWatcher) SetRefreshCoordinator(coordinator subscriptionRefr w.coordinatorMu.Lock() defer w.coordinatorMu.Unlock() if coordinator != nil && w.streamsConfigured { - for _, stream := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} { + for _, stream := range w.streamNames { coordinator.SetStreamCovered(stream, w.streams[stream].covered) } - for _, stream := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} { + for _, stream := range w.streamNames { state := w.streams[stream] coordinator.SetStreamLive(stream, state.live, state.live) } @@ -268,64 +313,75 @@ func (w *SubscriptionWatcher) handleSubscriptionLifecycle(stream RefreshStream, } } -func (w *SubscriptionWatcher) initializeStreamCoverage(topologyCovered bool) { +func (w *SubscriptionWatcher) initializeStreamCoverage(coverage map[RefreshStream]bool) { w.coordinatorMu.Lock() defer w.coordinatorMu.Unlock() w.streamsConfigured = true - w.streams[RefreshStreamTopology] = subscriptionWatcherStreamState{covered: topologyCovered} - w.streams[RefreshStreamDelivery] = subscriptionWatcherStreamState{covered: true} + for _, stream := range w.streamNames { + w.streams[stream] = subscriptionWatcherStreamState{covered: coverage[stream]} + } if w.coordinator == nil { return } - w.coordinator.SetStreamLive(RefreshStreamTopology, false, false) - w.coordinator.SetStreamLive(RefreshStreamDelivery, false, false) - w.coordinator.SetStreamCovered(RefreshStreamTopology, topologyCovered) - w.coordinator.SetStreamCovered(RefreshStreamDelivery, true) + // Fresh coverage supersedes any terminal degradation from a previous run. + w.coordinator.SetDegradedReason("") + for _, stream := range w.streamNames { + w.coordinator.SetStreamLive(stream, false, false) + } + for _, stream := range w.streamNames { + w.coordinator.SetStreamCovered(stream, coverage[stream]) + } } func (w *SubscriptionWatcher) clearStreamCoverage() { w.coordinatorMu.Lock() defer w.coordinatorMu.Unlock() w.streamsConfigured = false - w.streams[RefreshStreamTopology] = subscriptionWatcherStreamState{} - w.streams[RefreshStreamDelivery] = subscriptionWatcherStreamState{} + for _, stream := range w.streamNames { + w.streams[stream] = subscriptionWatcherStreamState{} + } if w.coordinator == nil { return } - w.coordinator.SetStreamLive(RefreshStreamTopology, false, false) - w.coordinator.SetStreamLive(RefreshStreamDelivery, false, false) - w.coordinator.SetStreamCovered(RefreshStreamTopology, false) - w.coordinator.SetStreamCovered(RefreshStreamDelivery, false) -} - -// Start begins subscribing to DWN record changes. It subscribes to all -// records under the wireguard-mesh protocol within the network context. -// -// The subscription automatically reconnects on failure (handled by -// SubscriptionManager). When any record changes, it triggers an -// immediate engine re-poll. -func (w *SubscriptionWatcher) Start(ctx context.Context) error { - w.mu.Lock() - defer w.mu.Unlock() - - if w.cancel != nil { - return nil // already started + for _, stream := range w.streamNames { + w.coordinator.SetStreamLive(stream, false, false) } - if w.selfDID == "" { - return fmt.Errorf("subscription watcher requires a self DID") + for _, stream := range w.streamNames { + w.coordinator.SetStreamCovered(stream, false) } +} - subCtx, cancel := context.WithCancel(ctx) - done := make(chan struct{}) - manager := w.newManager(w.endpoint, w.logger) +// setStreamUncovered drops one stream out of coverage after a terminal +// authorization failure and records the operator-facing reason. The +// coordinator falls back to the polling cadence. +func (w *SubscriptionWatcher) setStreamUncovered(stream RefreshStream, reason string) { + w.coordinatorMu.Lock() + defer w.coordinatorMu.Unlock() + state := w.streams[stream] + state.covered = false + state.live = false + w.streams[stream] = state + if w.coordinator == nil { + return + } + w.coordinator.SetStreamLive(stream, false, false) + w.coordinator.SetStreamCovered(stream, false) + w.coordinator.SetDegradedReason(reason) +} +// startRecordsSubscriptions opens the two Records streams: the broad +// mesh-protocol topology stream and the recipient-scoped delivery stream. +func (w *SubscriptionWatcher) startRecordsSubscriptions(subCtx context.Context, manager subscriptionManager) error { // Subscribe to all records under the wireguard-mesh protocol for // this network. This catches node, endpoint, and relay // record changes in a single subscription. Read grants authorize // RecordsSubscribe broadly. A role-only invocation cannot use this filter: // production DWN requires protocolPath and direct-parent scoping. meshAuth, topologySupported := broadSubscriptionAuth(w.readAuth) - w.initializeStreamCoverage(topologySupported) + w.initializeStreamCoverage(map[RefreshStream]bool{ + RefreshStreamTopology: topologySupported, + RefreshStreamDelivery: true, + }) if topologySupported { filter := dwn.RecordsFilter{ Protocol: protocols.MeshProtocolURI, @@ -344,10 +400,6 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { w.handleSubscriptionLifecycle(RefreshStreamTopology, event) }, ); err != nil { - w.clearStreamCoverage() - cancel() - manager.CloseAll() - w.clearStreamCoverage() return fmt.Errorf("subscribing to mesh records: %w", err) } } else { @@ -384,11 +436,113 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { w.handleSubscriptionLifecycle(RefreshStreamDelivery, event) }, ); err != nil { + return fmt.Errorf("subscribing to delivery records: %w", err) + } + return nil +} + +// startFeedSubscription opens the single Messages replication-feed stream. +// The server's core-protocol shadow filters make a protocol-scoped feed +// subscription also carry the $encryption/delivery control records and the +// mesh-tagged permission records, so this one stream subsumes both Records +// streams. Events remain invalidation-only wake hints. +func (w *SubscriptionWatcher) startFeedSubscription(subCtx context.Context, manager subscriptionManager) error { + // The feed stream is always authorized: the anchor owner bypasses grants + // and delegates hold protocol-root Messages.Read grants (feed mode is not + // selected otherwise). + w.initializeStreamCoverage(map[RefreshStream]bool{RefreshStreamFeed: true}) + + opts := dwn.MessagesSubscribeOptions{ + Filters: []dwn.MessagesFilter{{Protocol: protocols.MeshProtocolURI}}, + PermissionGrantIDs: append([]string(nil), w.feedGrantIDs...), + Cursor: w.feedSync.resumeCursor(), + } + if _, err := manager.SubscribeMessagesWithLifecycle( + subCtx, + w.anchorTenant, + w.signer, + opts, + w.handleFeedMessage, + w.handleFeedLifecycle, + ); err != nil { + return fmt.Errorf("subscribing to replication feed: %w", err) + } + return nil +} + +// handleFeedMessage records cursor progress for resume and translates feed +// events into invalidations. Events are never checkpoint evidence for the +// control-plane state — only the cursor advances; repairs still require a +// full refresh. +func (w *SubscriptionWatcher) handleFeedMessage(message *dwn.SubscriptionMessage) error { + if message == nil { + return nil + } + switch message.Type { + case dwn.SubscriptionEventType: + w.feedSync.recordCursor(message.Cursor) + case dwn.SubscriptionEOSEType: + w.feedSync.flushCursor(message.Cursor) + } + return w.handleSubscriptionMessage(RefreshStreamFeed, message) +} + +func (w *SubscriptionWatcher) handleFeedLifecycle(event dwn.SubscriptionLifecycleEvent) { + switch event.Kind { + case dwn.SubscriptionLifecycleProgressGap: + // The server can no longer resume the persisted cursor. Discard it so + // the reconnect replays from scratch; the lifecycle handler below + // holds the repair until the fresh stream is live. + w.feedSync.clearCursor() + case dwn.SubscriptionLifecycleTerminal: + if errors.Is(event.Err, dwn.ErrSubscriptionAuthorizationRevoked) { + // Grants revoked or expired mid-stream: the subscription cannot be + // re-established with the current session. Drop to uncovered + // polling fallback and surface the reason to `meshd status`. + w.logger.Warn("feed subscription authorization revoked; falling back to polling", + slog.Any("error", event.Err), + ) + w.setStreamUncovered(RefreshStreamFeed, feedAuthDegradedReason) + w.notify(RefreshReasonFeed) + return + } + } + w.handleSubscriptionLifecycle(RefreshStreamFeed, event) +} + +// Start begins subscribing to DWN record changes. It subscribes to all +// records under the wireguard-mesh protocol within the network context. +// +// The subscription automatically reconnects on failure (handled by +// SubscriptionManager). When any record changes, it triggers an +// immediate engine re-poll. +func (w *SubscriptionWatcher) Start(ctx context.Context) error { + w.mu.Lock() + defer w.mu.Unlock() + + if w.cancel != nil { + return nil // already started + } + if w.selfDID == "" { + return fmt.Errorf("subscription watcher requires a self DID") + } + + subCtx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + manager := w.newManager(w.endpoint, w.logger) + + var subscribeErr error + if w.feedMode { + subscribeErr = w.startFeedSubscription(subCtx, manager) + } else { + subscribeErr = w.startRecordsSubscriptions(subCtx, manager) + } + if subscribeErr != nil { w.clearStreamCoverage() cancel() manager.CloseAll() w.clearStreamCoverage() - return fmt.Errorf("subscribing to delivery records: %w", err) + return subscribeErr } w.cancel = cancel diff --git a/internal/engine/subscribe_test.go b/internal/engine/subscribe_test.go index 5d7f6b1..05851f4 100644 --- a/internal/engine/subscribe_test.go +++ b/internal/engine/subscribe_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "log/slog" "reflect" "sync" @@ -13,6 +14,7 @@ import ( "github.com/enboxorg/meshd/internal/dwn" dwncrypto "github.com/enboxorg/meshd/internal/dwn/crypto" + "github.com/enboxorg/meshd/internal/state" "github.com/enboxorg/meshd/protocols" "github.com/enboxorg/meshnet/control/controlclient" @@ -29,12 +31,21 @@ type recordedSubscription struct { lifecycle dwn.SubscriptionLifecycleHandler } +type recordedMessagesSubscription struct { + target string + signer *dwn.Signer + opts dwn.MessagesSubscribeOptions + handler dwn.EventHandler + lifecycle dwn.SubscriptionLifecycleHandler +} + type recordingSubscriptionManager struct { - mu sync.Mutex - calls []recordedSubscription - failCall int - closeHook func() - closeCall atomic.Int32 + mu sync.Mutex + calls []recordedSubscription + messagesCalls []recordedMessagesSubscription + failCall int + closeHook func() + closeCall atomic.Int32 } func (m *recordingSubscriptionManager) SubscribeWithAuthAndLifecycle( @@ -48,7 +59,7 @@ func (m *recordingSubscriptionManager) SubscribeWithAuthAndLifecycle( ) (*dwn.Subscription, error) { m.mu.Lock() defer m.mu.Unlock() - callNumber := len(m.calls) + 1 + callNumber := len(m.calls) + len(m.messagesCalls) + 1 if m.failCall == callNumber { return nil, errors.New("subscribe failed") } @@ -64,6 +75,30 @@ func (m *recordingSubscriptionManager) SubscribeWithAuthAndLifecycle( return nil, nil } +func (m *recordingSubscriptionManager) SubscribeMessagesWithLifecycle( + _ context.Context, + target string, + signer *dwn.Signer, + opts dwn.MessagesSubscribeOptions, + handler dwn.EventHandler, + lifecycle dwn.SubscriptionLifecycleHandler, +) (*dwn.Subscription, error) { + m.mu.Lock() + defer m.mu.Unlock() + callNumber := len(m.calls) + len(m.messagesCalls) + 1 + if m.failCall == callNumber { + return nil, errors.New("subscribe failed") + } + m.messagesCalls = append(m.messagesCalls, recordedMessagesSubscription{ + target: target, + signer: signer, + opts: opts, + handler: handler, + lifecycle: lifecycle, + }) + return nil, nil +} + func (m *recordingSubscriptionManager) CloseAll() { if m.closeHook != nil { m.closeHook() @@ -78,6 +113,7 @@ type refreshCoordinatorCall struct { live bool needsFullRefresh bool reason RefreshReason + degradedReason string } type recordingRefreshCoordinator struct { @@ -128,6 +164,12 @@ func (c *recordingRefreshCoordinator) Notify(reason RefreshReason) { c.calls = append(c.calls, refreshCoordinatorCall{method: "notify", reason: reason}) } +func (c *recordingRefreshCoordinator) SetDegradedReason(reason string) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls = append(c.calls, refreshCoordinatorCall{method: "degraded", degradedReason: reason}) +} + func (c *recordingRefreshCoordinator) takeCalls() []refreshCoordinatorCall { c.mu.Lock() defer c.mu.Unlock() @@ -298,7 +340,7 @@ func TestSubscriptionWatcherReplacementCoordinatorReplaysLiveCoverage(t *testing w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) first := newRecordingRefreshCoordinator() w.SetRefreshCoordinator(first) - w.initializeStreamCoverage(true) + w.initializeStreamCoverage(map[RefreshStream]bool{RefreshStreamTopology: true, RefreshStreamDelivery: true}) w.handleSubscriptionLifecycle(RefreshStreamTopology, dwn.SubscriptionLifecycleEvent{ Kind: dwn.SubscriptionLifecycleEstablished, }) @@ -321,7 +363,7 @@ func TestSubscriptionWatcherReplacementHandoffDoesNotOverwriteConcurrentDisconne w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) first := newRecordingRefreshCoordinator() w.SetRefreshCoordinator(first) - w.initializeStreamCoverage(true) + w.initializeStreamCoverage(map[RefreshStream]bool{RefreshStreamTopology: true, RefreshStreamDelivery: true}) w.handleSubscriptionLifecycle(RefreshStreamTopology, dwn.SubscriptionLifecycleEvent{ Kind: dwn.SubscriptionLifecycleEstablished, }) @@ -370,7 +412,7 @@ func TestSubscriptionWatcherEventIsFencedByCoordinatorHandoff(t *testing.T) { w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) first := newBlockingInvalidationCoordinator() w.SetRefreshCoordinator(first) - w.initializeStreamCoverage(true) + w.initializeStreamCoverage(map[RefreshStream]bool{RefreshStreamTopology: true, RefreshStreamDelivery: true}) w.handleSubscriptionLifecycle(RefreshStreamTopology, dwn.SubscriptionLifecycleEvent{ Kind: dwn.SubscriptionLifecycleEstablished, }) @@ -568,6 +610,7 @@ func TestSubscriptionWatcherStartUsesReadGrantAndRecipientDeliveryFilter(t *test } defer w.Stop() requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{ + {method: "degraded"}, {method: "live", stream: RefreshStreamTopology}, {method: "live", stream: RefreshStreamDelivery}, {method: "covered", stream: RefreshStreamTopology, covered: true}, @@ -634,6 +677,7 @@ func TestSubscriptionWatcherRoleOnlyUsesDeliveryAndPolling(t *testing.T) { } defer w.Stop() requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{ + {method: "degraded"}, {method: "live", stream: RefreshStreamTopology}, {method: "live", stream: RefreshStreamDelivery}, {method: "covered", stream: RefreshStreamTopology, covered: false}, @@ -971,6 +1015,193 @@ func TestSubscriptionWatcherStartWaitsForConcurrentStop(t *testing.T) { } } +func newTestFeedWatcher(manager subscriptionManager, coordinator subscriptionRefreshCoordinator, feedSync *feedSyncRecorder, grantIDs []string) *SubscriptionWatcher { + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ + AnchorEndpoint: "https://example.com", + AnchorTenant: "did:dht:anchor", + NetworkRecordID: "network-record", + SelfDID: "did:jwk:self", + Signer: &dwn.Signer{DID: "did:jwk:delegate"}, + FeedMode: true, + FeedGrantIDs: grantIDs, + }) + w.feedSync = feedSync + if manager != nil { + w.newManager = func(_ string, _ *slog.Logger) subscriptionManager { return manager } + } + if coordinator != nil { + w.SetRefreshCoordinator(coordinator) + } + return w +} + +func TestSubscriptionWatcherFeedModeStartsSingleFeedSubscription(t *testing.T) { + manager := &recordingSubscriptionManager{} + coordinator := newRecordingRefreshCoordinator() + feedSync := newFeedSyncRecorder(testFeedLinkKey, nil, nil) + feedSync.adopt(&state.FeedSyncState{LinkKey: testFeedLinkKey, Cursor: testCursor("5")}) + w := newTestFeedWatcher(manager, coordinator, feedSync, []string{"grant-a", "grant-b"}) + + if err := w.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + defer w.Stop() + + requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{ + {method: "degraded"}, + {method: "live", stream: RefreshStreamFeed}, + {method: "covered", stream: RefreshStreamFeed, covered: true}, + }) + + if len(manager.calls) != 0 { + t.Fatalf("records subscriptions = %d, want none in feed mode", len(manager.calls)) + } + if len(manager.messagesCalls) != 1 { + t.Fatalf("messages subscriptions = %d, want 1", len(manager.messagesCalls)) + } + call := manager.messagesCalls[0] + if call.target != "did:dht:anchor" { + t.Fatalf("target = %q, want anchor tenant", call.target) + } + wantFilters := []dwn.MessagesFilter{{Protocol: protocols.MeshProtocolURI}} + if !reflect.DeepEqual(call.opts.Filters, wantFilters) { + t.Fatalf("filters = %#v, want %#v", call.opts.Filters, wantFilters) + } + if !reflect.DeepEqual(call.opts.PermissionGrantIDs, []string{"grant-a", "grant-b"}) { + t.Fatalf("grant IDs = %v, want [grant-a grant-b]", call.opts.PermissionGrantIDs) + } + if call.opts.Cursor == nil || call.opts.Cursor.Position != "5" { + t.Fatalf("cursor = %+v, want resumed position 5", call.opts.Cursor) + } + if call.handler == nil || call.lifecycle == nil { + t.Fatal("feed handlers were not installed") + } +} + +func TestSubscriptionWatcherFeedModeOwnerOmitsCursorAndGrants(t *testing.T) { + manager := &recordingSubscriptionManager{} + // A checkpoint recorded under a different link key must not resume. + feedSync := newFeedSyncRecorder(testFeedLinkKey, nil, nil) + feedSync.adopt(&state.FeedSyncState{LinkKey: "stale^link^key^x", Cursor: testCursor("5")}) + w := newTestFeedWatcher(manager, newRecordingRefreshCoordinator(), feedSync, nil) + + if err := w.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + defer w.Stop() + + call := manager.messagesCalls[0] + if len(call.opts.PermissionGrantIDs) != 0 { + t.Fatalf("grant IDs = %v, want none for owner", call.opts.PermissionGrantIDs) + } + if call.opts.Cursor != nil { + t.Fatalf("cursor = %+v, want nil after link key mismatch", call.opts.Cursor) + } +} + +func TestSubscriptionWatcherFeedEventsInvalidateAndPersistCursor(t *testing.T) { + saves := &feedSaveRecorder{} + feedSync := newFeedSyncRecorder(testFeedLinkKey, saves.save, nil) + feedSync.debounce = time.Hour // only EOSE persists synchronously here + coordinator := newRecordingRefreshCoordinator() + w := newTestFeedWatcher(nil, coordinator, feedSync, nil) + + if err := w.handleFeedMessage(&dwn.SubscriptionMessage{ + Type: dwn.SubscriptionEventType, + Cursor: testCursor("7"), + }); err != nil { + t.Fatalf("handleFeedMessage event: %v", err) + } + requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{{ + method: "invalidate", stream: RefreshStreamFeed, reason: RefreshReasonFeed, + }}) + if got := saves.takeSaves(); len(got) != 0 { + t.Fatalf("event persisted synchronously: %+v", got) + } + if got := feedSync.resumeCursor(); got == nil || got.Position != "7" { + t.Fatalf("recorded cursor = %+v, want position 7", got) + } + + // EOSE is a barrier: cursor persists immediately, no invalidation. + if err := w.handleFeedMessage(&dwn.SubscriptionMessage{ + Type: dwn.SubscriptionEOSEType, + Cursor: testCursor("9"), + }); err != nil { + t.Fatalf("handleFeedMessage eose: %v", err) + } + requireRefreshCoordinatorCalls(t, coordinator, nil) + got := saves.takeSaves() + if len(got) != 1 || got[0].Cursor == nil || got[0].Cursor.Position != "9" { + t.Fatalf("eose saves = %+v, want one save at position 9", got) + } +} + +func TestSubscriptionWatcherFeedProgressGapClearsCursor(t *testing.T) { + saves := &feedSaveRecorder{} + feedSync := newFeedSyncRecorder(testFeedLinkKey, saves.save, nil) + feedSync.flushCursor(testCursor("7")) + saves.takeSaves() + coordinator := newRecordingRefreshCoordinator() + w := newTestFeedWatcher(nil, coordinator, feedSync, nil) + + w.handleFeedLifecycle(dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleProgressGap, + Gap: &dwn.ProgressGapInfo{Reason: "epoch_mismatch"}, + }) + + requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{{ + method: "live", stream: RefreshStreamFeed, needsFullRefresh: true, + }}) + got := saves.takeSaves() + if len(got) != 1 || got[0].Cursor != nil { + t.Fatalf("gap saves = %+v, want one save with a cleared cursor", got) + } + if feedSync.resumeCursor() != nil { + t.Fatal("cursor survived a progress gap") + } +} + +func TestSubscriptionWatcherFeedTerminalAuthDegradesToPolling(t *testing.T) { + coordinator := newRecordingRefreshCoordinator() + w := newTestFeedWatcher(nil, coordinator, nil, nil) + w.initializeStreamCoverage(map[RefreshStream]bool{RefreshStreamFeed: true}) + coordinator.takeCalls() + + authErr := fmt.Errorf("terminal subscription error: %s: revoked: %w", + dwn.MessagesSubscribeDeliveryAuthorizationFailedCode, dwn.ErrSubscriptionAuthorizationRevoked) + w.handleFeedLifecycle(dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleTerminal, + Err: authErr, + }) + + requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{ + {method: "live", stream: RefreshStreamFeed}, + {method: "covered", stream: RefreshStreamFeed}, + {method: "degraded", degradedReason: feedAuthDegradedReason}, + {method: "notify", reason: RefreshReasonFeed}, + }) + if health := coordinator.streamHealth(RefreshStreamFeed); health.Covered || health.Live { + t.Fatalf("feed stream health = %+v, want uncovered", health) + } +} + +func TestSubscriptionWatcherFeedTerminalWithoutAuthSentinelKeepsCoverage(t *testing.T) { + coordinator := newRecordingRefreshCoordinator() + w := newTestFeedWatcher(nil, coordinator, nil, nil) + w.initializeStreamCoverage(map[RefreshStream]bool{RefreshStreamFeed: true}) + coordinator.takeCalls() + + w.handleFeedLifecycle(dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleTerminal, + Err: errors.New("connection torn down"), + }) + + requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{ + {method: "live", stream: RefreshStreamFeed}, + {method: "notify", reason: RefreshReasonFeed}, + }) +} + func TestEngineCreatesSubscriptionWatcher(t *testing.T) { cfg := Config{ AnchorEndpoint: "https://example.com", @@ -995,4 +1226,50 @@ func TestEngineCreatesSubscriptionWatcher(t *testing.T) { if eng.subWatcher.readAuth.ProtocolRole != "network/node" { t.Fatalf("watcher read role = %q, want network/node", eng.subWatcher.readAuth.ProtocolRole) } + // A role-only node is not feed-eligible: records streams, no recorder. + if eng.subWatcher.feedMode || eng.feedSync != nil { + t.Fatalf("feedMode = %v, feedSync = %v; want records mode", eng.subWatcher.feedMode, eng.feedSync) + } +} + +func TestEngineOwnerRunsFeedModeAndAdoptsCheckpoint(t *testing.T) { + cfg := Config{ + AnchorEndpoint: "https://example.com", + AnchorTenant: "did:dht:owner", + NetworkRecordID: "record123", + SelfDID: "did:dht:owner", + Signer: &dwn.Signer{DID: "did:dht:owner"}, + } + linkKey, err := feedLinkKey(cfg) + if err != nil { + t.Fatalf("feedLinkKey: %v", err) + } + saves := &feedSaveRecorder{} + cfg.FeedSyncLoad = func() (*state.FeedSyncState, error) { + return &state.FeedSyncState{ + LinkKey: linkKey, + Cursor: testCursor("11"), + Fingerprint: "ff00", + }, nil + } + cfg.FeedSyncSave = saves.save + + eng, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + defer eng.Stop() + + if !eng.subWatcher.feedMode { + t.Fatal("owner engine should run in feed mode") + } + if eng.feedSync == nil || eng.subWatcher.feedSync != eng.feedSync { + t.Fatal("feed sync recorder not shared between engine and watcher") + } + if got := eng.feedSync.resumeCursor(); got == nil || got.Position != "11" { + t.Fatalf("resume cursor = %+v, want adopted position 11", got) + } + if got := eng.feedSync.storedFingerprint(); got != "ff00" { + t.Fatalf("stored fingerprint = %q, want adopted ff00", got) + } } From 2760edbff53237d23ef2bc27f03155808a204c83 Mon Sep 17 00:00:00 2001 From: Liran Cohen Date: Sun, 26 Jul 2026 15:54:16 +0000 Subject: [PATCH 3/4] fix(mesh,cmd): wire feed mode through meshd up and surface degraded status DelegateSession now resolves the active protocol-root Messages.Read grants for the mesh protocol into canonical grant IDs plus the {id, dateGranted, dateExpires} epoch tuples; sessions without feed coverage stay on the Records streams. cmd/meshd threads them (and the feedsync.json load/save hooks) into engine.Config like the existing grant plumbing, and clears the feed checkpoint on network leave. daemon.SnapshotStatus carries the coordinator's DegradedReason so 'meshd status' can tell operators when wallet session grants expired mid-stream and a reconnect is needed. Refs #267 Co-Authored-By: Claude Fable 5 --- cmd/meshd/main.go | 20 ++++ cmd/meshd/peer_list_local_test.go | 31 ++++++ internal/daemon/daemon.go | 5 + internal/mesh/delegate.go | 42 ++++++++ internal/mesh/delegate_feed_test.go | 145 ++++++++++++++++++++++++++++ 5 files changed, 243 insertions(+) create mode 100644 internal/mesh/delegate_feed_test.go diff --git a/cmd/meshd/main.go b/cmd/meshd/main.go index 7e0f0d9..d2352e8 100644 --- a/cmd/meshd/main.go +++ b/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,6 +4226,14 @@ 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 @@ -4224,6 +4241,8 @@ func cmdUp(ctx context.Context, args []string, flagProfile string) error { 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 diff --git a/cmd/meshd/peer_list_local_test.go b/cmd/meshd/peer_list_local_test.go index a01b354..1b2c672 100644 --- a/cmd/meshd/peer_list_local_test.go +++ b/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) { diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 64b7fda..2905632 100644 --- a/internal/daemon/daemon.go +++ b/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 diff --git a/internal/mesh/delegate.go b/internal/mesh/delegate.go index de1ef17..cf39d96 100644 --- a/internal/mesh/delegate.go +++ b/internal/mesh/delegate.go @@ -51,6 +51,16 @@ type DelegateSession struct { WriteGrant json.RawMessage DeleteGrant json.RawMessage + // FeedGrantIDs are the active protocol-root Messages.Read grant IDs + // covering the mesh protocol's replication feed, in canonical invocation + // order. Empty when the wallet session lacks feed coverage — the engine + // then keeps the two Records subscription streams. + FeedGrantIDs []string + + // FeedGrants are the {id, dateGranted, dateExpires} tuples of + // FeedGrantIDs, hashed into the feed link's authorization epoch. + FeedGrants []dwn.SyncAuthorizationGrant + logger *slog.Logger } @@ -99,6 +109,8 @@ func NewDelegateSession(ctx context.Context, params DelegateSessionParams) (*Del return nil, fmt.Errorf("wallet session is missing delegated read/write grants for %s", protocols.MeshProtocolURI) } + s.FeedGrantIDs, s.FeedGrants = resolveFeedGrants(params.Grants, params.OwnerDID, s.DelegateDID, now, logger) + client := dwn.NewClient(params.Endpoint, params.DelegateSigner) def, err := FetchInstalledProtocolDefinition(ctx, client, params.OwnerDID, protocols.MeshProtocolURI) @@ -129,6 +141,36 @@ func NewDelegateSession(ctx context.Context, params DelegateSessionParams) (*Del return s, nil } +// resolveFeedGrants selects the active protocol-root Messages.Read grants +// that authorize replication-feed reads (MessagesQuery/Subscribe) of the mesh +// protocol on the owner tenant. Sessions without feed coverage return empty +// results and the caller stays on the Records subscription streams; feed mode +// is an upgrade, never a requirement. +func resolveFeedGrants(rawGrants []json.RawMessage, ownerDID, delegateDID string, now time.Time, logger *slog.Logger) ([]string, []dwn.SyncAuthorizationGrant) { + parsed := make([]*dwn.PermissionGrant, 0, len(rawGrants)) + for _, raw := range rawGrants { + grant, err := dwn.ParsePermissionGrant(raw) + if err != nil { + // Sessions may carry non-grant records; skip them. + continue + } + parsed = append(parsed, grant) + } + + scope, err := dwn.ProtocolSetSyncScope([]string{protocols.MeshProtocolURI}) + if err != nil { + return nil, nil + } + selected, err := dwn.SelectFeedGrants(parsed, ownerDID, delegateDID, now, scope) + if err != nil { + logger.Info("wallet session has no Messages.Read feed coverage; keeping Records subscription streams", + slog.Any("error", err), + ) + return nil, nil + } + return dwn.FeedGrantIDs(selected), dwn.SyncAuthorizationGrantsFromPermissionGrants(selected) +} + // ReadAuth returns the message auth for delegated read/query operations. func (s *DelegateSession) ReadAuth() dwn.MessageAuth { return dwn.MessageAuth{DelegatedGrant: s.ReadGrant} diff --git a/internal/mesh/delegate_feed_test.go b/internal/mesh/delegate_feed_test.go new file mode 100644 index 0000000..be7d03c --- /dev/null +++ b/internal/mesh/delegate_feed_test.go @@ -0,0 +1,145 @@ +package mesh + +import ( + "encoding/base64" + "encoding/json" + "log/slog" + "reflect" + "testing" + "time" + + "github.com/enboxorg/meshd/internal/dwn" + "github.com/enboxorg/meshd/protocols" +) + +const ( + feedTestOwner = "did:dht:owner" + feedTestDelegate = "did:jwk:delegate" +) + +var feedTestNow = time.Date(2026, 7, 26, 12, 0, 0, 0, time.UTC) + +// feedTestGrantMessage builds a raw grant message in the shape +// dwn.ParsePermissionGrant expects (mirrors the wallet's grant records). +func feedTestGrantMessage(t *testing.T, id string, scope dwn.PermissionScope, expires string) json.RawMessage { + t.Helper() + header, err := json.Marshal(map[string]string{"alg": "EdDSA", "kid": feedTestOwner + "#0"}) + if err != nil { + t.Fatal(err) + } + data, err := json.Marshal(map[string]any{ + "dateExpires": expires, + "scope": scope, + }) + if err != nil { + t.Fatal(err) + } + msg, err := json.Marshal(map[string]any{ + "recordId": id, + "encodedData": base64.RawURLEncoding.EncodeToString(data), + "descriptor": map[string]any{ + "recipient": feedTestDelegate, + "dateCreated": "2026-07-01T00:00:00.000000Z", + }, + "authorization": map[string]any{ + "signature": map[string]any{ + "payload": "", + "signatures": []map[string]any{{ + "protected": base64.RawURLEncoding.EncodeToString(header), + "signature": "", + }}, + }, + }, + }) + if err != nil { + t.Fatal(err) + } + return msg +} + +func messagesReadScope(protocol string) dwn.PermissionScope { + return dwn.PermissionScope{ + Interface: dwn.DwnScopeInterfaceMessages, + Method: dwn.DwnScopeMethodRead, + Protocol: protocol, + } +} + +func TestResolveFeedGrantsSelectsProtocolRootMessagesRead(t *testing.T) { + grants := []json.RawMessage{ + // Records grants never authorize feed scopes. + feedTestGrantMessage(t, "grant-records-read", dwn.PermissionScope{ + Interface: dwn.DwnScopeInterfaceRecords, + Method: dwn.DwnScopeMethodRead, + Protocol: protocols.MeshProtocolURI, + }, "2026-08-01T00:00:00.000000Z"), + // Two protocol-root Messages.Read grants participate, sorted by ID. + feedTestGrantMessage(t, "grant-b", messagesReadScope(protocols.MeshProtocolURI), "2026-08-01T00:00:00.000000Z"), + feedTestGrantMessage(t, "grant-a", messagesReadScope(protocols.MeshProtocolURI), "2026-08-15T00:00:00.000000Z"), + // Expired grants are excluded. + feedTestGrantMessage(t, "grant-expired", messagesReadScope(protocols.MeshProtocolURI), "2026-07-01T00:00:00.000000Z"), + // Path-scoped Messages grants cover a strict subset and are excluded. + feedTestGrantMessage(t, "grant-scoped", dwn.PermissionScope{ + Interface: dwn.DwnScopeInterfaceMessages, + Method: dwn.DwnScopeMethodRead, + Protocol: protocols.MeshProtocolURI, + ProtocolPath: "network/node", + }, "2026-08-01T00:00:00.000000Z"), + // Non-grant session records are skipped. + json.RawMessage(`{"kind":"not-a-grant"}`), + } + + ids, tuples := resolveFeedGrants(grants, feedTestOwner, feedTestDelegate, feedTestNow, slog.Default()) + if !reflect.DeepEqual(ids, []string{"grant-a", "grant-b"}) { + t.Fatalf("feed grant IDs = %v, want [grant-a grant-b]", ids) + } + if len(tuples) != 2 { + t.Fatalf("feed grant tuples = %+v, want 2", tuples) + } + if tuples[0].ID != "grant-a" || tuples[0].DateExpires != "2026-08-15T00:00:00.000000Z" || + tuples[0].DateGranted != "2026-07-01T00:00:00.000000Z" { + t.Fatalf("tuple[0] = %+v, want grant-a epoch fields", tuples[0]) + } + if tuples[1].ID != "grant-b" || tuples[1].DateExpires != "2026-08-01T00:00:00.000000Z" { + t.Fatalf("tuple[1] = %+v, want grant-b epoch fields", tuples[1]) + } +} + +func TestResolveFeedGrantsWithoutCoverageReturnsEmpty(t *testing.T) { + tests := []struct { + name string + grants []json.RawMessage + }{ + {name: "no grants"}, + { + name: "records grants only", + grants: []json.RawMessage{ + feedTestGrantMessage(t, "grant-records", dwn.PermissionScope{ + Interface: dwn.DwnScopeInterfaceRecords, + Method: dwn.DwnScopeMethodRead, + Protocol: protocols.MeshProtocolURI, + }, "2026-08-01T00:00:00.000000Z"), + }, + }, + { + name: "messages grant for a different protocol", + grants: []json.RawMessage{ + feedTestGrantMessage(t, "grant-other", messagesReadScope("https://example.com/protocols/other"), "2026-08-01T00:00:00.000000Z"), + }, + }, + { + name: "expired messages grant", + grants: []json.RawMessage{ + feedTestGrantMessage(t, "grant-expired", messagesReadScope(protocols.MeshProtocolURI), "2026-07-01T00:00:00.000000Z"), + }, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ids, tuples := resolveFeedGrants(test.grants, feedTestOwner, feedTestDelegate, feedTestNow, slog.Default()) + if len(ids) != 0 || len(tuples) != 0 { + t.Fatalf("resolveFeedGrants = %v/%v, want empty (records mode fallback)", ids, tuples) + } + }) + } +} From 43e9d0cda5c5d4afd2ec237fb1e7875cf2ff2cda Mon Sep 17 00:00:00 2001 From: Liran Cohen Date: Sun, 26 Jul 2026 16:10:33 +0000 Subject: [PATCH 4/4] fix(dwn): reserve auth-revoked sentinel for authorization failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MessagesSubscribeDeliveryFailed is the server's generic non-DwnError delivery fault, unrelated to grants, but it was wrapped in ErrSubscriptionAuthorizationRevoked alongside the authorization code. That made handleFeedLifecycle drop the feed stream to uncovered polling and surface the wallet-grant degraded reason ('run meshd auth connect') — categorically wrong on anchor-owner nodes, which have no wallet session, and misleading for delegates. Only MessagesSubscribeDeliveryAuthorizationFailed now wraps the sentinel; DeliveryFailed takes the plain terminal path (coverage retained, live=false, fallback cadence), matching the enbox reference agent, which pauses only on auth codes and repairs/retries otherwise. Refs #267 Co-Authored-By: Claude Fable 5 --- internal/dwn/messages_subscribe_test.go | 2 +- internal/dwn/subscribe.go | 15 ++++---- internal/engine/subscribe_test.go | 48 ++++++++++++++++++------- 3 files changed, 44 insertions(+), 21 deletions(-) diff --git a/internal/dwn/messages_subscribe_test.go b/internal/dwn/messages_subscribe_test.go index 244cdfe..09d6eb9 100644 --- a/internal/dwn/messages_subscribe_test.go +++ b/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}, } diff --git a/internal/dwn/subscribe.go b/internal/dwn/subscribe.go index 0d22f4e..42fbe63 100644 --- a/internal/dwn/subscribe.go +++ b/internal/dwn/subscribe.go @@ -131,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. @@ -804,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: diff --git a/internal/engine/subscribe_test.go b/internal/engine/subscribe_test.go index 05851f4..1c5940a 100644 --- a/internal/engine/subscribe_test.go +++ b/internal/engine/subscribe_test.go @@ -1186,20 +1186,42 @@ func TestSubscriptionWatcherFeedTerminalAuthDegradesToPolling(t *testing.T) { } func TestSubscriptionWatcherFeedTerminalWithoutAuthSentinelKeepsCoverage(t *testing.T) { - coordinator := newRecordingRefreshCoordinator() - w := newTestFeedWatcher(nil, coordinator, nil, nil) - w.initializeStreamCoverage(map[RefreshStream]bool{RefreshStreamFeed: true}) - coordinator.takeCalls() - - w.handleFeedLifecycle(dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleTerminal, - Err: errors.New("connection torn down"), - }) + tests := []struct { + name string + err error + }{ + {name: "generic transport error", err: errors.New("connection torn down")}, + { + // A server-side delivery fault is not a grant problem: it must + // not trip the auth-degraded reason (which prescribes a wallet + // remedy inapplicable on owner nodes). + name: "delivery failed terminal", + err: fmt.Errorf("terminal subscription error: %s: %s", + dwn.MessagesSubscribeDeliveryFailedCode, "delivery stopped"), + }, + } - requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{ - {method: "live", stream: RefreshStreamFeed}, - {method: "notify", reason: RefreshReasonFeed}, - }) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + coordinator := newRecordingRefreshCoordinator() + w := newTestFeedWatcher(nil, coordinator, nil, nil) + w.initializeStreamCoverage(map[RefreshStream]bool{RefreshStreamFeed: true}) + coordinator.takeCalls() + + w.handleFeedLifecycle(dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleTerminal, + Err: tc.err, + }) + + requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{ + {method: "live", stream: RefreshStreamFeed}, + {method: "notify", reason: RefreshReasonFeed}, + }) + if health := coordinator.streamHealth(RefreshStreamFeed); !health.Covered { + t.Fatalf("feed stream health = %+v, want covered", health) + } + }) + } } func TestEngineCreatesSubscriptionWatcher(t *testing.T) {