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/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/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/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 212556a..42fbe63 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 @@ -124,10 +131,10 @@ const ( ) // ErrSubscriptionAuthorizationRevoked marks a terminal subscription error -// caused by the server permanently stopping delivery — grant revocation, -// grant expiry, or an unrecoverable delivery failure. The terminal lifecycle -// error wraps this sentinel so callers can distinguish "re-resolve the grant -// set" from other terminal conditions via errors.Is. +// caused by delivery-time grant revalidation failing — grant revocation or +// grant expiry. The terminal lifecycle error wraps this sentinel so callers +// can distinguish "re-resolve the grant set" from other terminal conditions +// (including generic delivery failures) via errors.Is. var ErrSubscriptionAuthorizationRevoked = errors.New("subscription delivery authorization revoked") // ProgressGapInfo describes why a stored cursor cannot be resumed. @@ -797,9 +804,10 @@ func (s *Subscription) handleSubscriptionFrame(ctx context.Context, conn *websoc var terminalErr error if messageType == SubscriptionErrorType { switch message.Error.Code { - case MessagesSubscribeDeliveryAuthorizationFailedCode, MessagesSubscribeDeliveryFailedCode: - // Grant revocation/expiry (or unrecoverable delivery failure): - // wrap the sentinel so callers can react specifically. + case MessagesSubscribeDeliveryAuthorizationFailedCode: + // Grant revocation/expiry: wrap the sentinel so callers can react + // specifically. DeliveryFailed is a server-side delivery fault, not + // a grant problem, so it takes the plain terminal path below. terminalErr = fmt.Errorf("terminal subscription error: %s: %s: %w", message.Error.Code, message.Error.Detail, ErrSubscriptionAuthorizationRevoked) default: 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..1c5940a 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,215 @@ 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) { + 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"), + }, + } + + 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) { cfg := Config{ AnchorEndpoint: "https://example.com", @@ -995,4 +1248,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) + } } 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) + } + }) + } +} 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) + } +}