Skip to content

Commit 262aff6

Browse files
authored
Merge pull request #392 from Toluwalase-Daniel/fix/issues-352-353-354-355
Fix non-atomic eventStore/webhookRepository writes and O(n) recipient upserts (#352, #353, #354, #355)
2 parents 40b5e6a + 3fb7139 commit 262aff6

5 files changed

Lines changed: 391 additions & 39 deletions

File tree

‎src/indexer/eventStore.js‎

Lines changed: 71 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -83,8 +83,43 @@ async function upsertAirdrop(event) {
8383
});
8484
}
8585

86-
await cache.set(airdropKey(airdropId), next);
87-
await cache.getClient().sadd(AIRDROP_IDS_KEY, airdropId);
86+
// Issue #354: a single MULTI/EXEC transaction instead of two separate
87+
// round trips — a crash between them used to leave an airdrop record
88+
// with no entry in AIRDROP_IDS_KEY (invisible to anything that scans the
89+
// id set) or vice versa (an id with no backing record).
90+
await cache.getClient().multi()
91+
.set(airdropKey(airdropId), JSON.stringify(next))
92+
.sadd(AIRDROP_IDS_KEY, airdropId)
93+
.exec();
94+
}
95+
96+
// Issue #353: recipients used to be stored as one JSON-list string per
97+
// airdrop, so every single-recipient upsert had to load the ENTIRE list
98+
// into memory (O(n) per event) just to update one entry. Each recipient is
99+
// now its own field in a Redis hash keyed by airdrop id, so an upsert is
100+
// one HGET + one HSET (O(1)), regardless of how many recipients the
101+
// airdrop has.
102+
//
103+
// A key that predates this fix still holds the old JSON-list string, and
104+
// HGET/HSET on a String-typed key raises WRONGTYPE — this migrates it,
105+
// once, lazily on first access, rather than requiring a manual operator
106+
// step before deploying.
107+
async function migrateRecipientsListToHashIfNeeded(key) {
108+
const redis = cache.getClient();
109+
const type = await redis.type(key);
110+
if (type !== 'string') return;
111+
112+
const list = await getJsonList(key);
113+
await redis.del(key);
114+
if (list.length === 0) return;
115+
116+
const pipeline = redis.pipeline();
117+
for (const entry of list) {
118+
if (entry && entry.recipient) {
119+
pipeline.hset(key, entry.recipient, JSON.stringify(entry));
120+
}
121+
}
122+
await pipeline.exec();
88123
}
89124

90125
async function upsertRecipient(event) {
@@ -93,9 +128,11 @@ async function upsertRecipient(event) {
93128
if (!airdropId || !recipient) return;
94129

95130
const key = recipientsKey(airdropId);
96-
const recipients = await getJsonList(key);
97-
const existingIndex = recipients.findIndex((entry) => entry.recipient === recipient);
98-
const existing = existingIndex >= 0 ? recipients[existingIndex] : { recipient };
131+
await migrateRecipientsListToHashIfNeeded(key);
132+
133+
const redis = cache.getClient();
134+
const existingRaw = await redis.hget(key, recipient);
135+
const existing = existingRaw ? JSON.parse(existingRaw) : { recipient };
99136
const next = {
100137
...existing,
101138
airdrop_id: airdropId,
@@ -115,10 +152,7 @@ async function upsertRecipient(event) {
115152
next.claimed_at = event.ledger_closed_at;
116153
}
117154

118-
if (existingIndex >= 0) recipients[existingIndex] = next;
119-
else recipients.push(next);
120-
121-
await setJsonList(key, recipients);
155+
await redis.hset(key, recipient, JSON.stringify(next));
122156
}
123157

124158
async function appendClaim(event) {
@@ -142,8 +176,18 @@ async function appendClaim(event) {
142176
}
143177

144178
async function saveEvent(event) {
145-
await cache.set(eventKey(event.id), event);
146-
await cache.getClient().sadd(EVENT_IDS_KEY, event.id);
179+
// Issue #352: the raw event record and its id-set membership are written
180+
// in one MULTI/EXEC transaction — a crash between them used to leave an
181+
// event id in EVENT_IDS_KEY with no backing record, or a record with no
182+
// id-set entry (invisible to getEventCount()). upsertAirdrop/
183+
// upsertRecipient/appendClaim below each read-then-write their own
184+
// separate keys (their prior values determine what gets written), so
185+
// they can't join this same transaction — see #354 and #353 for their
186+
// own atomicity/scalability fixes.
187+
await cache.getClient().multi()
188+
.set(eventKey(event.id), JSON.stringify(event))
189+
.sadd(EVENT_IDS_KEY, event.id)
190+
.exec();
147191
await upsertAirdrop(event);
148192
await upsertRecipient(event);
149193
await appendClaim(event);
@@ -191,7 +235,22 @@ async function getAirdropStatus(airdropId) {
191235
}
192236

193237
async function getAirdropRecipients(airdropId) {
194-
return getJsonList(recipientsKey(airdropId));
238+
const key = recipientsKey(airdropId);
239+
await migrateRecipientsListToHashIfNeeded(key);
240+
241+
const redis = cache.getClient();
242+
const raw = await redis.hgetall(key);
243+
const recipients = Object.values(raw).map((v) => JSON.parse(v));
244+
// A hash has no guaranteed iteration order across calls; sort by
245+
// added_ledger (falling back to recipient address) so pagination here
246+
// stays stable, matching the old JSON-list's insertion order.
247+
recipients.sort((a, b) => {
248+
const la = a.added_ledger ?? 0;
249+
const lb = b.added_ledger ?? 0;
250+
if (la !== lb) return la - lb;
251+
return String(a.recipient).localeCompare(String(b.recipient));
252+
});
253+
return recipients;
195254
}
196255

197256
async function getRecipientClaims(address) {

‎src/repositories/webhookRepository.js‎

Lines changed: 17 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -85,15 +85,25 @@ async function create({ url, events, secret, description, filters, owner_ip }) {
8585
updated_at: now,
8686
};
8787
const redis = cache.getClient();
88-
await cache.set(key(id), record);
89-
// A sorted set scored by creation time, not a plain set — mirrors
90-
// airdropsService/alerts.js's own IDS_KEY pattern, so paginating (added
91-
// below) walks a deterministic, newest-first order rather than
92-
// whatever arbitrary order SMEMBERS happened to return (#131).
93-
await redis.zadd(IDS_KEY, Date.parse(now), id);
88+
// Issue #355: one MULTI/EXEC transaction instead of two or three separate
89+
// round trips — every value here is freshly computed above (no read of
90+
// prior state feeds into any of these writes), so unlike the read-modify-
91+
// write helpers in eventStore.js, this can go directly into one atomic
92+
// batch. A crash mid-way used to leave a webhook record with no entry in
93+
// IDS_KEY (invisible to list()/listAll(), so it would never fire), or an
94+
// id in IDS_KEY with no backing record (a 404 from findById() that list()
95+
// still returned as a phantom entry).
96+
const multi = redis.multi()
97+
.set(key(id), JSON.stringify(record))
98+
// A sorted set scored by creation time, not a plain set — mirrors
99+
// airdropsService/alerts.js's own IDS_KEY pattern, so paginating (added
100+
// below) walks a deterministic, newest-first order rather than
101+
// whatever arbitrary order SMEMBERS happened to return (#131).
102+
.zadd(IDS_KEY, Date.parse(now), id);
94103
if (owner_ip) {
95-
await redis.zadd(`webhooks:owner:${owner_ip}`, Date.parse(now), id);
104+
multi.zadd(`webhooks:owner:${owner_ip}`, Date.parse(now), id);
96105
}
106+
await multi.exec();
97107
return normalize(record);
98108
}
99109

‎test/helpers/cacheMock.js‎

Lines changed: 121 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -2,17 +2,20 @@
22

33
/**
44
* In-memory mock of the ioredis surface used by src/services/cache.js.
5-
* Covers strings (used by cache.get/set/del), SETs, sorted SETs, and LISTs.
5+
* Covers strings (used by cache.get/set/del), SETs, sorted SETs, LISTs,
6+
* and HASHes.
67
*/
78
function createCacheMock() {
8-
const store = new Map();
99
const sets = new Map();
1010
const zsets = new Map();
1111
const lists = new Map();
12+
const hashes = new Map();
1213
const counters = new Map();
13-
// Separate raw string store (with per-key TTL) backing redis.set/get/del —
14-
// distinct from `store` above, which backs the higher-level cacheMock.
15-
// get/set JSON API with different key/value semantics.
14+
// Raw string store (with per-key TTL) backing redis.set/get/del. Real
15+
// Redis has exactly one string keyspace behind both the raw ioredis API
16+
// and cache.js's JSON-serializing get/set/del wrapper — so both are
17+
// implemented against this single Map (see cacheMock.get/set/del below),
18+
// rather than each holding its own disconnected copy of the same key.
1619
const rawStore = new Map();
1720

1821
function getSet(key) {
@@ -27,6 +30,10 @@ function createCacheMock() {
2730
if (!lists.has(key)) lists.set(key, []);
2831
return lists.get(key);
2932
}
33+
function getHash(key) {
34+
if (!hashes.has(key)) hashes.set(key, new Map());
35+
return hashes.get(key);
36+
}
3037
function isExpired(entry) {
3138
return entry.expiresAt !== null && Date.now() >= entry.expiresAt;
3239
}
@@ -36,6 +43,38 @@ function createCacheMock() {
3643
return entry;
3744
}
3845

46+
// Queues commands issued through redis.multi()/redis.pipeline() and
47+
// replays them, in order, against the same redis.<command> mock
48+
// functions on .exec() — so MULTI/EXEC and PIPELINE go through the exact
49+
// same logic as calling each command directly, with no duplicated
50+
// implementation to drift out of sync.
51+
function makeChain() {
52+
const queue = [];
53+
const chain = {};
54+
const proxy = new Proxy(chain, {
55+
get(target, prop) {
56+
if (prop === 'exec') {
57+
return async () => {
58+
const results = [];
59+
for (const [cmd, args] of queue) {
60+
try {
61+
results.push([null, await redis[cmd](...args)]);
62+
} catch (err) {
63+
results.push([err, null]);
64+
}
65+
}
66+
return results;
67+
};
68+
}
69+
return (...args) => {
70+
queue.push([prop, args]);
71+
return proxy;
72+
};
73+
},
74+
});
75+
return proxy;
76+
}
77+
3978
const redis = {
4079
smembers: jest.fn(async (key) => [...(sets.get(key) || [])]),
4180
sadd: jest.fn(async (key, val) => { getSet(key).add(val); }),
@@ -113,19 +152,63 @@ function createCacheMock() {
113152
}
114153
if (nx && getLive(key)) return null;
115154
rawStore.set(key, { value: String(value), expiresAt: ttlMs !== null ? Date.now() + ttlMs : null });
155+
hashes.delete(key);
156+
sets.delete(key);
157+
zsets.delete(key);
158+
lists.delete(key);
116159
return 'OK';
117160
}),
118161
get: jest.fn(async (key) => {
119162
const entry = getLive(key);
120163
return entry ? entry.value : null;
121164
}),
122-
del: jest.fn(async (key) => (rawStore.delete(key) ? 1 : 0)),
165+
del: jest.fn(async (key) => {
166+
const had = getLive(key) !== null || hashes.has(key) || sets.has(key) || zsets.has(key) || lists.has(key);
167+
rawStore.delete(key);
168+
hashes.delete(key);
169+
sets.delete(key);
170+
zsets.delete(key);
171+
lists.delete(key);
172+
return had ? 1 : 0;
173+
}),
174+
// Issue #353 (eventStore.js) uses these to store per-recipient fields
175+
// instead of one JSON-list string per airdrop.
176+
type: jest.fn(async (key) => {
177+
if (getLive(key)) return 'string';
178+
if (hashes.has(key) && hashes.get(key).size > 0) return 'hash';
179+
if (sets.has(key) && sets.get(key).size > 0) return 'set';
180+
if (zsets.has(key) && zsets.get(key).size > 0) return 'zset';
181+
if (lists.has(key) && lists.get(key).length > 0) return 'list';
182+
return 'none';
183+
}),
184+
hset: jest.fn(async (key, field, value) => {
185+
const isNew = !getHash(key).has(field);
186+
getHash(key).set(field, String(value));
187+
return isNew ? 1 : 0;
188+
}),
189+
hget: jest.fn(async (key, field) => {
190+
const h = hashes.get(key);
191+
return h && h.has(field) ? h.get(field) : null;
192+
}),
193+
hgetall: jest.fn(async (key) => {
194+
const h = hashes.get(key);
195+
if (!h) return {};
196+
return Object.fromEntries(h.entries());
197+
}),
123198
pexpire: jest.fn(async (key, ms) => {
124199
const entry = getLive(key);
125200
if (!entry) return 0;
126201
entry.expiresAt = Date.now() + Number(ms);
127202
return 1;
128203
}),
204+
// Issues #352/#354/#355: MULTI/EXEC — queues commands and replays them
205+
// against the mock's own already-implemented methods on .exec(). Real
206+
// MULTI/EXEC additionally guarantees no other client's commands can
207+
// interleave between them; this mock only needs to prove the caller
208+
// issues one .exec() covering all the intended writes; it does not
209+
// model cross-client interleaving.
210+
multi: jest.fn(() => makeChain()),
211+
pipeline: jest.fn(() => makeChain()),
129212
// Mimics ioredis#defineCommand for the custom commands this codebase
130213
// registers (see deliveryRepository.js and leaderElection.js). Real
131214
// Redis runs the Lua body single-threaded to completion, so this mock
@@ -180,19 +263,45 @@ function createCacheMock() {
180263
getCommandQueueLength: () => 0,
181264
getConcurrencyStats: () => ({ active: 0, waiting: 0, available: 50, max: 50 }),
182265
get: jest.fn(async (key) => {
183-
const v = store.get(key);
184-
return v !== undefined ? JSON.parse(JSON.stringify(v)) : null;
266+
const raw = await redis.get(key);
267+
if (raw === null || raw === undefined) return null;
268+
try {
269+
return JSON.parse(raw);
270+
} catch {
271+
return null;
272+
}
185273
}),
186-
set: jest.fn(async (key, value) => { store.set(key, JSON.parse(JSON.stringify(value))); }),
187-
del: jest.fn(async (key) => { store.delete(key); }),
274+
set: jest.fn(async (key, value, ttlSeconds) => {
275+
const serialized = JSON.stringify(value);
276+
if (ttlSeconds) await redis.set(key, serialized, 'EX', ttlSeconds);
277+
else await redis.set(key, serialized);
278+
}),
279+
del: jest.fn(async (key) => { await redis.del(key); }),
188280
disconnect: jest.fn(async () => {}),
189281
};
190282

283+
// Read-only view over rawStore, parsed, for tests that inspect the
284+
// stored value directly rather than through cache.get() (e.g.
285+
// webhookRepository.test.js's encryption-at-rest assertions). Backed by
286+
// the same rawStore cache.set/get and raw redis.set/get share, so it
287+
// sees a write made through either API.
288+
const store = {
289+
get(key) {
290+
const entry = getLive(key);
291+
if (!entry) return undefined;
292+
try {
293+
return JSON.parse(entry.value);
294+
} catch {
295+
return entry.value;
296+
}
297+
},
298+
};
299+
191300
function reset() {
192-
store.clear();
193301
sets.clear();
194302
zsets.clear();
195303
lists.clear();
304+
hashes.clear();
196305
counters.clear();
197306
rawStore.clear();
198307
Object.values(redis).forEach((fn) => fn.mockClear?.());
@@ -201,7 +310,7 @@ function createCacheMock() {
201310
cacheMock.del.mockClear();
202311
}
203312

204-
return { cacheMock, redis, store, sets, zsets, lists, counters, reset };
313+
return { cacheMock, redis, store, sets, zsets, lists, hashes, counters, reset };
205314
}
206315

207316
module.exports = { createCacheMock };

0 commit comments

Comments
 (0)