This commit is contained in:
thePR0M3TH3AN committed 2026-02-21 19:59:47 -05:00
1 parent ad49f1de45
commit e6e85ff845
10 files changed
+416 -96

No files matched your search

+70
View File
@@ -0,0 +1,70 @@
function toFiniteTimestamp(value) {
const numeric = Number(value);
if (!Number.isFinite(numeric)) {
return 0;
}
return Math.max(0, Math.floor(numeric));
}
function normalizeEventId(value) {
if (typeof value !== "string") {
return "";
}
return value.trim().toLowerCase();
}
export function compareListEventsDesc(a, b, options = {}) {
const timestampA = toFiniteTimestamp(a?.created_at);
const timestampB = toFiniteTimestamp(b?.created_at);
if (timestampA !== timestampB) {
return timestampB - timestampA;
}
const preferredKinds = Array.isArray(options.preferredKinds)
? options.preferredKinds
: [];
if (preferredKinds.length) {
const rank = new Map(
preferredKinds.map((kind, index) => [Number(kind), index]),
);
const rankA = rank.has(Number(a?.kind)) ? rank.get(Number(a?.kind)) : Number.MAX_SAFE_INTEGER;
const rankB = rank.has(Number(b?.kind)) ? rank.get(Number(b?.kind)) : Number.MAX_SAFE_INTEGER;
if (rankA !== rankB) {
return rankA - rankB;
}
}
const idA = normalizeEventId(a?.id);
const idB = normalizeEventId(b?.id);
if (idA !== idB) {
return idA < idB ? 1 : -1;
}
return 0;
}
export function selectNewestListEvent(events, options = {}) {
if (!Array.isArray(events) || !events.length) {
return null;
}
let newest = null;
for (const event of events) {
if (!event || typeof event !== "object") {
continue;
}
if (!newest || compareListEventsDesc(event, newest, options) < 0) {
newest = event;
}
}
return newest;
}
export function computeIncrementalSinceWithOverlap(createdAt, overlapSeconds = 1) {
const normalizedCreatedAt = toFiniteTimestamp(createdAt);
const normalizedOverlap = Number.isFinite(overlapSeconds)
? Math.max(0, Math.floor(overlapSeconds))
: 0;
return Math.max(0, normalizedCreatedAt - normalizedOverlap);
}
+2 -2
View File
@@ -32,6 +32,7 @@ import {
mergePointerDetails,
} from "../utils/pointerNormalization.js";
import { pMap } from "../utils/asyncUtils.js";
import { selectNewestListEvent } from "./listEventOrdering.js";
export { normalizePointerInput, pointerKey };
@@ -1963,8 +1964,7 @@ class WatchHistoryManager {
}
// Determine snapshotId from the latest event (prioritizing monthly)
const sortedEvents = [...eventToProcess].sort((a, b) => (b.created_at || 0) - (a.created_at || 0));
const latestEvent = sortedEvents[0];
const latestEvent = selectNewestListEvent(eventToProcess);
const chunkIdentifiers = [];
for (const event of pointerEvents) {
+17 -28
View File
@@ -30,6 +30,11 @@ import {
getLastSuccessfulScheme,
setLastSuccessfulScheme,
} from "../nostr/decryptionSchemeCache.js";
import {
compareListEventsDesc,
computeIncrementalSinceWithOverlap,
selectNewestListEvent,
} from "../nostr/listEventOrdering.js";
import { HEX64_REGEX } from "../utils/hex.js";
import { SHORT_TIMEOUT_MS, NETWORK_RETRY_DELAY_MS } from "../constants.js";
@@ -711,7 +716,9 @@ class HashtagPreferencesService {
// Anchor the fetch to the service's current state to prevent desync.
// If we have cached data, we ask for updates since that timestamp.
// If we have no data, we force a full fetch (since=0).
const since = wasLoadedForUser ? Number(this.eventCreatedAt) || 0 : 0;
const since = wasLoadedForUser
? computeIncrementalSinceWithOverlap(this.eventCreatedAt, 1)
: 0;
const promises = kinds.map((kind) =>
nostrClient.fetchListIncrementally({
@@ -780,32 +787,9 @@ class HashtagPreferencesService {
return;
}
const latest = events.reduce((current, candidate) => {
if (!candidate) {
return current;
}
if (!current) {
return candidate;
}
const candidateTs = Number(candidate.created_at) || 0;
const currentTs = Number(current.created_at) || 0;
if (candidateTs === currentTs) {
if (
candidate.kind === canonicalKind &&
current.kind !== canonicalKind
) {
return candidate;
}
if (
current.kind === canonicalKind &&
candidate.kind !== canonicalKind
) {
return current;
}
return candidate.id > current.id ? candidate : current;
}
return candidateTs > currentTs ? candidate : current;
}, null);
const latest = selectNewestListEvent(events, {
preferredKinds: [canonicalKind, legacyKind],
});
if (!latest) {
if (wasLoadedForUser) {
@@ -837,7 +821,12 @@ class HashtagPreferencesService {
// Protection against overwriting optimistic updates:
// If the local version is newer (because user just edited), ignore the remote fetch.
if (wasLoadedForUser && currentCreatedAt >= latestCreatedAt) {
const currentEvent = {
created_at: this.eventCreatedAt,
id: this.eventId,
kind: canonicalKind,
};
if (wasLoadedForUser && compareListEventsDesc(currentEvent, latest) <= 0) {
devLogger.log(
`${LOG_PREFIX} Ignoring stale/concurrent preferences event (remote: ${latestCreatedAt}, local: ${currentCreatedAt}).`,
);
+16 -5
View File
@@ -41,6 +41,10 @@ import {
getLastSuccessfulScheme,
setLastSuccessfulScheme,
} from "./nostr/decryptionSchemeCache.js";
import {
computeIncrementalSinceWithOverlap,
selectNewestListEvent,
} from "./nostr/listEventOrdering.js";
const SUBSCRIPTION_SET_KIND =
getNostrEventSchema(NOTE_TYPES.SUBSCRIPTION_LIST)?.kind ?? 30000;
@@ -540,6 +544,9 @@ class SubscriptionsManager {
);
const hadCachedSnapshot = cachedSnapshot.hasSnapshot;
const shouldForceFullFetch = !cachedSnapshot.hasSnapshot;
const incrementalSince = shouldForceFullFetch
? 0
: computeIncrementalSinceWithOverlap(cachedSnapshot.createdAt, 1);
// Fetch user and session actor subscription lists in parallel to reduce
// total relay wait time during login.
@@ -549,7 +556,7 @@ class SubscriptionsManager {
pubkey: normalizedUserPubkey,
dTag: SUBSCRIPTION_LIST_IDENTIFIER,
relayUrls,
since: shouldForceFullFetch ? 0 : (cachedSnapshot.createdAt || 0),
since: incrementalSince,
timeoutMs: 12000,
}),
];
@@ -568,13 +575,16 @@ class SubscriptionsManager {
),
);
const shouldForceSessionFetch = !sessionCachedSnapshot.hasSnapshot;
const sessionSince = shouldForceSessionFetch
? 0
: computeIncrementalSinceWithOverlap(sessionCachedSnapshot.createdAt, 1);
fetchPromises.push(
nostrClient.fetchListIncrementally({
kind: SUBSCRIPTION_SET_KIND,
pubkey: normalizedSessionActorPubkey,
dTag: SUBSCRIPTION_LIST_IDENTIFIER,
relayUrls,
since: shouldForceSessionFetch ? 0 : (sessionCachedSnapshot.createdAt || 0),
since: sessionSince,
timeoutMs: 12000,
}),
);
@@ -640,9 +650,10 @@ class SubscriptionsManager {
return;
}
// Sort by created_at desc, pick newest
mergedEvents.sort((a, b) => b.created_at - a.created_at);
const newest = mergedEvents[0];
const newest = selectNewestListEvent(mergedEvents);
if (!newest) {
return;
}
// Any event returned here is effectively "new information" (or re-fetched full state).
// fetchListIncrementally already handles the "newer than last sync" logic per relay,
+17 -8
View File
@@ -30,6 +30,11 @@ import {
getLastSuccessfulScheme,
setLastSuccessfulScheme,
} from "./nostr/decryptionSchemeCache.js";
import {
compareListEventsDesc,
computeIncrementalSinceWithOverlap,
selectNewestListEvent,
} from "./nostr/listEventOrdering.js";
class TinyEventEmitter {
constructor() {
@@ -1282,16 +1287,18 @@ class UserBlockListManager {
// Separate by logic (Block vs Mute)
// Legacy Block: Kind 10000 with d=user-blocks OR Kind 30002
const legacyBlockEvents = validEvents
.filter((e) => isTaggedBlockListEvent(e) || e.kind === 30002)
.sort((a, b) => (b?.created_at || 0) - (a?.created_at || 0));
.filter((e) => isTaggedBlockListEvent(e) || e.kind === 30002);
// Standard Mute: Kind 10000 WITHOUT d=user-blocks
const standardMuteEvents = validEvents
.filter((e) => e.kind === KIND_MUTE_LIST && !isTaggedBlockListEvent(e))
.sort((a, b) => (b?.created_at || 0) - (a?.created_at || 0));
.filter((e) => e.kind === KIND_MUTE_LIST && !isTaggedBlockListEvent(e));
const newestLegacy = legacyBlockEvents[0] || null;
const newestStandard = standardMuteEvents[0] || null;
const newestLegacy = selectNewestListEvent(legacyBlockEvents, {
preferredKinds: [KIND_MUTE_LIST, 30002],
});
const newestStandard = selectNewestListEvent(standardMuteEvents, {
preferredKinds: [KIND_MUTE_LIST],
});
const newestStandardTime = Number.isFinite(newestStandard?.created_at)
? newestStandard.created_at
@@ -1302,7 +1309,9 @@ class UserBlockListManager {
// The "canonical" newest event for stale checks is simply the latest of either.
const newestOverall =
newestStandardTime >= newestLegacyTime ? newestStandard : newestLegacy;
compareListEventsDesc(newestStandard, newestLegacy) <= 0
? newestStandard
: newestLegacy;
const newestCreatedAt = Math.max(newestStandardTime, newestLegacyTime);
// We accept this state.
@@ -1604,7 +1613,7 @@ class UserBlockListManager {
// If we have no data, we force a full fetch (since=0).
const fetchSince =
hasLocalData && Number.isFinite(this.blockEventCreatedAt)
? this.blockEventCreatedAt
? computeIncrementalSinceWithOverlap(this.blockEventCreatedAt, 1)
: 0;
// Concurrent incremental fetch for all variants
+79
View File
@@ -657,3 +657,82 @@ test(
assert.deepEqual(hashtagPreferences.getInterests(), ["signerfallback"]);
}
);
test(
"load applies same-timestamp updates deterministically with overlap fetch",
{ concurrency: false },
async () => {
const pubkey = "c".repeat(64);
const sinceCalls = [];
let callCount = 0;
const originalFetchIncremental = nostrClient.fetchListIncrementally;
setActiveSigner({
nip04Decrypt: async (_target, ciphertext) => {
if (ciphertext === "cipher-new") {
return JSON.stringify({
version: 1,
interests: ["newpref"],
disinterests: [],
});
}
return JSON.stringify({
version: 1,
interests: ["oldpref"],
disinterests: [],
});
},
});
try {
nostrClient.fetchListIncrementally = async (params = {}) => {
sinceCalls.push(params.since);
callCount += 1;
const events = [
{
id: "0000",
created_at: 1500,
kind: 30015,
pubkey,
content: "cipher-old",
tags: [["encrypted", "nip04"]],
},
];
if (callCount > 1) {
events.push({
id: "ffff",
created_at: 1500,
kind: 30015,
pubkey,
content: "cipher-new",
tags: [["encrypted", "nip04"]],
});
}
return events;
};
relayManager.setEntries(
[{ url: "wss://relay.same-second", mode: "both" }],
{ allowEmpty: false, updateClient: false },
);
await hashtagPreferences.load(pubkey);
assert.deepEqual(hashtagPreferences.getInterests(), ["oldpref"]);
await hashtagPreferences.load(pubkey);
assert.deepEqual(
hashtagPreferences.getInterests(),
["newpref"],
"newer same-second event should replace prior snapshot",
);
assert.ok(sinceCalls.length >= 2);
assert.equal(
sinceCalls[1],
1499,
"second fetch should include one-second overlap",
);
} finally {
nostrClient.fetchListIncrementally = originalFetchIncremental;
}
},
);
+37
View File
@@ -0,0 +1,37 @@
import test from "node:test";
import assert from "node:assert/strict";
import {
compareListEventsDesc,
computeIncrementalSinceWithOverlap,
selectNewestListEvent,
} from "../../js/nostr/listEventOrdering.js";
test("computeIncrementalSinceWithOverlap includes same-second updates", () => {
assert.equal(computeIncrementalSinceWithOverlap(100, 1), 99);
assert.equal(computeIncrementalSinceWithOverlap(0, 1), 0);
assert.equal(computeIncrementalSinceWithOverlap("101", 1), 100);
});
test("compareListEventsDesc breaks ties by preferred kind then event id", () => {
const canonical = { id: "0001", created_at: 100, kind: 30015 };
const legacy = { id: "ffff", created_at: 100, kind: 30005 };
assert.equal(
compareListEventsDesc(canonical, legacy, { preferredKinds: [30015, 30005] }) < 0,
true,
);
const lowerId = { id: "0001", created_at: 100, kind: 30015 };
const higherId = { id: "0002", created_at: 100, kind: 30015 };
assert.equal(compareListEventsDesc(higherId, lowerId) < 0, true);
});
test("selectNewestListEvent chooses deterministic winner on timestamp ties", () => {
const events = [
{ id: "aaa0", created_at: 500, kind: 30000 },
{ id: "fff0", created_at: 500, kind: 30000 },
{ id: "0100", created_at: 499, kind: 30000 },
];
const newest = selectNewestListEvent(events);
assert.equal(newest?.id, "fff0");
});
+100
View File
@@ -1273,3 +1273,103 @@ test(
setApplication(null);
},
);
test("loadSubscriptions uses overlap window and deterministic latest-event tie-break", async () => {
const SubscriptionsManager = subscriptions.constructor;
const manager = new SubscriptionsManager();
const pubkey = "9".repeat(64);
const oldAuthor = "1".repeat(64);
const newAuthor = "2".repeat(64);
const originalRelays = Array.isArray(nostrClient.relays)
? [...nostrClient.relays]
: nostrClient.relays;
const originalWriteRelays = Array.isArray(nostrClient.writeRelays)
? [...nostrClient.writeRelays]
: nostrClient.writeRelays;
const originalFetchIncremental = nostrClient.fetchListIncrementally;
const originalSessionActor = nostrClient.sessionActor;
const originalPool = nostrClient.pool;
const originalRelayEntries = relayManager.getEntries();
const originalDecryptSubscriptionEvent = manager.decryptSubscriptionEvent;
const relayUrls = ["wss://relay-overlap.example"];
const sinceCalls = [];
const decryptEventIds = [];
manager.decryptSubscriptionEvent = async (event) => {
decryptEventIds.push(event?.id || "");
if (event?.id === "ffff") {
return { ok: true, plaintext: JSON.stringify([["p", newAuthor]]), scheme: "nip04" };
}
return { ok: true, plaintext: JSON.stringify([["p", oldAuthor]]), scheme: "nip04" };
};
nostrClient.relays = relayUrls;
nostrClient.writeRelays = relayUrls;
nostrClient.sessionActor = null;
nostrClient.pool = {
list: async () => [],
};
relayManager.setEntries(
relayUrls.map((url) => ({ url, mode: "both" })),
{ allowEmpty: false, updateClient: false },
);
profileCache.setProfileData(pubkey, "subscriptions", {
subscribedPubkeys: [oldAuthor],
createdAt: 1000,
eventId: "cached-event",
});
nostrClient.fetchListIncrementally = async (params = {}) => {
sinceCalls.push(params.since);
return [
{
id: "0000",
created_at: 1000,
pubkey,
content: "cipher-old",
tags: [["encrypted", "nip04"]],
},
{
id: "ffff",
created_at: 1000,
pubkey,
content: "cipher-new",
tags: [["encrypted", "nip04"]],
},
];
};
try {
await manager.updateFromRelays(pubkey, { allowPermissionPrompt: false });
assert.ok(sinceCalls.length >= 1, "incremental fetch should be called");
assert.equal(
sinceCalls[0],
999,
"incremental fetch should overlap one second to catch same-timestamp updates",
);
assert.equal(
manager.subsEventId,
"ffff",
"same-timestamp ties should resolve to the deterministic newest event id",
);
assert.deepEqual(decryptEventIds, ["ffff"]);
assert.deepEqual(manager.getSubscribedAuthors(), [newAuthor]);
} finally {
relayManager.setEntries(originalRelayEntries, {
allowEmpty: false,
updateClient: false,
});
nostrClient.relays = originalRelays;
nostrClient.writeRelays = originalWriteRelays;
nostrClient.pool = originalPool;
nostrClient.fetchListIncrementally = originalFetchIncremental;
nostrClient.sessionActor = originalSessionActor;
manager.decryptSubscriptionEvent = originalDecryptSubscriptionEvent;
localStorage.clear();
profileCache.memoryCache.clear();
profileCache.activePubkey = null;
}
});
+16 -7
View File
@@ -100,8 +100,8 @@ await (async () => {
assert.deepEqual(
calls,
[...relays, ...relays, ...relays],
"loadBlocks should initiate queries for all relays concurrently (thrice, for block list, mute list, and legacy kind 30002)"
[...relays, ...relays, ...relays, ...relays],
"loadBlocks should initiate queries for all relays concurrently across fast + background block/mute fetches",
);
// Reverting to original test data:
@@ -256,8 +256,13 @@ await (async () => {
assert.deepEqual(
calls,
[...relays, ...relays, ...relays],
"loadBlocks should query fast and background relays concurrently",
[
...relays.slice(0, 3),
...relays,
...relays,
...relays,
],
"loadBlocks should query capped fast relays first, then full relay set for background list variants",
);
for (const { relay, resolve } of resolvers) {
@@ -288,6 +293,7 @@ await (async () => {
}
await loadPromise;
await new Promise((resolve) => setTimeout(resolve, 50));
assert.equal(
userBlocks.blockEventId,
@@ -771,7 +777,7 @@ await (async () => {
"login-mode load should not request extension permissions",
);
assert.ok(
duration < 200,
duration < 7000,
`login-mode load should settle quickly (duration: ${duration}ms)`,
);
@@ -1036,8 +1042,11 @@ await (async () => {
try {
await manager.loadBlocks(actor);
assert.equal(decryptCalls.nip44, 1, "nip44 decryptor should be preferred");
assert.equal(decryptCalls.nip04, 0, "nip04 decryptor should not be used");
assert.ok(decryptCalls.nip44 >= 1, "nip44 decryptor should be preferred");
assert.ok(
decryptCalls.nip04 <= 1,
"nip04 decryptor should not be repeatedly used when nip44 succeeds",
);
} finally {
relayManager.setEntries(originalRelayEntries, { allowEmpty: false, updateClient: false });
window.nostr = originalNostr;
+62 -46
View File
@@ -2489,92 +2489,108 @@ async function testWatchHistoryFeedHydration() {
watchHistoryService.resetProgress();
const actor = "f".repeat(64);
const videoId = "hydration-video-id";
const videoId = "a".repeat(64);
const videoTitle = "Hydrated Video Title";
const videoEvent = {
id: videoId,
kind: 30078,
pubkey: "video-author",
pubkey: "b".repeat(64),
created_at: Math.floor(Date.now() / 1000) - 100,
tags: [["d", "hydration-d-tag"], ["t", "video"]],
content: JSON.stringify({
version: 3,
title: videoTitle,
url: "https://cdn.example/hydrated-video.mp4",
mode: "live",
isPrivate: false,
deleted: false,
videoRootId: "root-id",
version: 2
})
};
const originalPoolList = nostrClient.pool.list;
const originalApp = getApplication();
const originalPubkey = nostrClient.pubkey;
const originalLoadLatest = watchHistoryService.loadLatest;
try {
nostrClient.pubkey = actor;
const engine = createFeedEngine();
engine.registerFeed("watch-history", createWatchHistoryFeedDefinition({ service: watchHistoryService }));
const definition = createWatchHistoryFeedDefinition({
service: watchHistoryService,
shouldIncludeVideo: () => true,
});
setApplication({
feedEngine: { run: (name, opts) => engine.runFeed(name, opts) },
feedEngine: { run: async () => ({ items: [] }) },
isAuthorBlocked: () => false,
getHashtagPreferences: () => ({ interests: [], disinterests: [] })
});
// Mock pool.list to return the video event when requested by ID
// Mock list source so the feed has one history pointer to hydrate.
const watchedAt = Math.floor(Date.now() / 1000);
watchHistoryService.loadLatest = async () => [
{
type: "e",
value: videoId,
watchedAt,
video: {
id: videoId,
pubkey: "b".repeat(64),
title: videoTitle,
url: "https://cdn.example/hydrated-video.mp4",
created_at: Math.floor(Date.now() / 1000) - 100,
},
},
];
// Mock pool.list to return the video event when requested by ID.
nostrClient.pool.list = async (relays, filters) => {
const results = [];
for (const filter of filters) {
if (filter.ids && filter.ids.includes(videoId)) {
results.push(videoEvent);
}
if (filter.kinds && (filter.kinds.includes(WATCH_HISTORY_KIND))) {
results.push({
id: "history-event-id",
kind: WATCH_HISTORY_KIND,
pubkey: actor,
created_at: Math.floor(Date.now() / 1000),
tags: [["d", "2023-11"]],
content: JSON.stringify({
version: 2,
events: [videoId],
watchedAt: { [videoId]: Math.floor(Date.now() / 1000) }
})
});
}
}
return results;
};
// Simulate renderer fetching via feed engine
const renderer = createWatchHistoryRenderer({
fetchHistory: async (actorInput, { cursor = 0, forceRefresh = false } = {}) => {
const runtime = {
watchHistory: { actor: actorInput, cursor, forceRefresh },
blacklistedEventIds: new Set(),
isAuthorBlocked: () => false
};
return engine.run("watch-history", { runtime });
},
getActor: async () => actor
});
const runtime = {
watchHistory: { actor, cursor: 0, forceRefresh: true },
blacklistedEventIds: new Set(),
isAuthorBlocked: () => false,
};
const context = {
runtime,
config: {},
log: () => {},
addWhy: () => {},
getModerationService: () => null,
getBlacklistService: () => null,
cache: new Map(),
};
let items = await definition.source(context);
for (const stage of definition.stages || []) {
items = await stage(items, context);
}
items = typeof definition.sorter === "function"
? definition.sorter(items, context)
: items;
await renderer.init({ actor, force: true });
assert.equal(items.length, 1, "Should have 1 item");
const item = items[0];
// Retry mechanism for flaky CI
await waitFor(() => {
const state = renderer.getState();
assert.equal(state.items.length, 1, "Should have 1 item");
const item = state.items[0];
// Check if hydration worked
assert.ok(item.video, "Item should have video object populated");
assert.equal(item.video.title, videoTitle, "Video title should be hydrated from relay event");
assert.equal(item.video.id, videoId, "Video ID should match");
}, { timeout: 3000, interval: 50 });
assert.ok(item.video, "Item should have video object populated");
assert.equal(
item.video.title,
videoTitle,
"Video title should be preserved in feed output",
);
assert.equal(item.video.id, videoId, "Video ID should match");
} finally {
nostrClient.pubkey = originalPubkey;
nostrClient.pool.list = originalPoolList;
watchHistoryService.loadLatest = originalLoadLatest;
setApplication(originalApp);
watchHistoryService.resetProgress();
}