Skip to content

feat: add bypassDataItemFilter flag and convert to object parameters (PE-8173) #433

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

feat: add bypassDataItemFilter flag and convert to object parameters (PE-8173) #433

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions src/lib/ans-104.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,11 +281,13 @@ export class Ans104Parser {
parentId,
parentIndex,
rootParentOffset,
bypassDataItemFilter = false,
}: {
rootTxId: string;
parentId: string;
parentIndex: number;
rootParentOffset: number;
bypassDataItemFilter?: boolean;
}): Promise<void> {
// eslint-disable-next-line no-async-promise-executor
return new Promise(async (resolve, reject) => {
Expand Down Expand Up @@ -347,6 +349,7 @@ export class Ans104Parser {
parentIndex,
bundlePath,
rootParentOffset,
bypassDataItemFilter,
},
});
this.drainQueue();
Expand Down Expand Up @@ -421,8 +424,14 @@ if (!isMainThread) {
process.exit(0);
}

const { rootTxId, parentId, parentIndex, bundlePath, rootParentOffset } =
message;
const {
rootTxId,
parentId,
parentIndex,
bundlePath,
rootParentOffset,
bypassDataItemFilter = false,
} = message;
let stream: fs.ReadStream | undefined = undefined;
try {
stream = fs.createReadStream(bundlePath);
Expand Down Expand Up @@ -478,7 +487,7 @@ if (!isMainThread) {
rootParentOffset,
});

if (await filter.match(normalizedDataItem)) {
if (bypassDataItemFilter || (await filter.match(normalizedDataItem))) {
matchedItemCount++;
parentPort?.postMessage({
eventName: DATA_ITEM_MATCHED,
Expand Down
49 changes: 39 additions & 10 deletions src/routes/ar-io.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -257,20 +257,46 @@ arIoRouter.post(
express.json(),
async (req, res) => {
try {
const { id, bypassFilter = true } = req.body;
const {
id,
bypassBundleFilter = true,
bypassDataItemFilter = false,
// Legacy support
bypassFilter,
} = req.body;

if (id === undefined) {
res.status(400).send("Must provide 'id'");
return;
}

if (bypassFilter !== undefined && typeof bypassFilter !== 'boolean') {
res.status(400).send("'bypassFilter' must be a boolean");
// Handle legacy bypassFilter parameter
const effectiveBypassBundleFilter =
bypassFilter !== undefined ? bypassFilter : bypassBundleFilter;

if (
effectiveBypassBundleFilter !== undefined &&
typeof effectiveBypassBundleFilter !== 'boolean'
) {
// Use legacy error message if using legacy parameter
const errorMessage =
bypassFilter !== undefined
? "'bypassFilter' must be a boolean"
: "'bypassBundleFilter' must be a boolean";
res.status(400).send(errorMessage);
return;
}

if (
bypassDataItemFilter !== undefined &&
typeof bypassDataItemFilter !== 'boolean'
) {
res.status(400).send("'bypassDataItemFilter' must be a boolean");
return;
}

// if byPassFilter is false, then queue like queue-tx
if (bypassFilter === false) {
// if bypassBundleFilter is false, then queue like queue-tx
if (effectiveBypassBundleFilter === false) {
system.prioritizedTxIds.add(id);
system.txFetcher.queueTxId({ txId: id });
res.json({ message: 'TX queued' });
Expand All @@ -284,11 +310,14 @@ arIoRouter.post(
return;
}

const queuedBundle = await system.queueBundle(
{ id, root_tx_id: id } as NormalizedDataItem | PartialJsonTransaction,
true,
bypassFilter,
);
const queuedBundle = await system.queueBundle({
item: { id, root_tx_id: id } as
| NormalizedDataItem
| PartialJsonTransaction,
prioritized: true,
bypassBundleFilter: effectiveBypassBundleFilter,
bypassDataItemFilter,
});

if (queuedBundle.error !== undefined) {
res.status(503).send(queuedBundle.error);
Expand Down
38 changes: 25 additions & 13 deletions src/system.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -611,11 +611,19 @@ export type QueueBundleResponse = {
status: 'skipped' | 'queued' | 'error';
error?: string;
};
export async function queueBundle(
item: NormalizedDataItem | PartialJsonTransaction,
isPrioritized = false,
bypassFilter = false,
): Promise<QueueBundleResponse> {
export interface QueueBundleOptions {
item: NormalizedDataItem | PartialJsonTransaction;
prioritized?: boolean;
bypassBundleFilter?: boolean;
bypassDataItemFilter?: boolean;
}

export async function queueBundle({
item,
prioritized = false,
bypassBundleFilter = false,
bypassDataItemFilter = false,
}: QueueBundleOptions): Promise<QueueBundleResponse> {
try {
if ('root_tx_id' in item && item.root_tx_id === null) {
log.debug('Skipping download of optimistically indexed data item', {
Expand All @@ -632,7 +640,10 @@ export async function queueBundle(
format: 'ans-104',
});

if (bypassFilter || (await config.ANS104_UNBUNDLE_FILTER.match(item))) {
if (
bypassBundleFilter ||
(await config.ANS104_UNBUNDLE_FILTER.match(item))
) {
metrics.bundlesMatchedCounter.inc({ bundle_format: 'ans-104' });
const {
unbundleFilterId,
Expand Down Expand Up @@ -662,17 +673,18 @@ export async function queueBundle(
return { status: 'skipped' };
}

bundleDataImporter.queueItem(
{
bundleDataImporter.queueItem({
item: {
...item,
index:
'parent_index' in item && item.parent_index !== undefined
? item.parent_index
: -1, // parent indexes are not needed for L1
},
isPrioritized,
bypassFilter,
);
prioritized,
bypassBundleFilter,
bypassDataItemFilter,
});
metrics.bundlesQueuedCounter.inc({ bundle_format: 'ans-104' });
} else {
await db.saveBundle({
Expand Down Expand Up @@ -701,15 +713,15 @@ eventEmitter.on(
const isPrioritized = prioritizedTxIds.has(item.id);
prioritizedTxIds.delete(item.id);

await queueBundle(item, isPrioritized);
await queueBundle({ item, prioritized: isPrioritized });
},
);

// Queue nested bundles
eventEmitter.on(
events.ANS104_NESTED_BUNDLE_INDEXED,
async (item: NormalizedDataItem | PartialJsonTransaction) => {
await queueBundle(item, true);
await queueBundle({ item, prioritized: true });
},
);

Expand Down
15 changes: 10 additions & 5 deletions src/workers/ans104-unbundler.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ describe('Ans104Unbundler', () => {
shouldUnbundleMock.mock.mockImplementation(() => false);

for (let i = 0; i < 10; i++) {
ans104Unbundler.queueItem(mockItem, false);
ans104Unbundler.queueItem({ item: mockItem, prioritized: false });
}

assert.equal(shouldUnbundleMock.mock.calls.length, 10);
Expand All @@ -77,7 +77,7 @@ describe('Ans104Unbundler', () => {

it('should queue item when shouldUnbundle returns true', async () => {
for (let i = 0; i < 10; i++) {
ans104Unbundler.queueItem(mockItem, false);
ans104Unbundler.queueItem({ item: mockItem, prioritized: false });
}

assert.equal(shouldUnbundleMock.mock.calls.length, 10);
Expand All @@ -88,7 +88,7 @@ describe('Ans104Unbundler', () => {
shouldUnbundleMock.mock.mockImplementation(() => false);

for (let i = 0; i < 10; i++) {
ans104Unbundler.queueItem(mockItem, true);
ans104Unbundler.queueItem({ item: mockItem, prioritized: true });
}

assert.equal(shouldUnbundleMock.mock.calls.length, 10);
Expand All @@ -99,7 +99,7 @@ describe('Ans104Unbundler', () => {
ans104Unbundler['workerCount'] = 0;

for (let i = 0; i < 10; i++) {
ans104Unbundler.queueItem(mockItem, false);
ans104Unbundler.queueItem({ item: mockItem, prioritized: false });
}

assert.equal(shouldUnbundleMock.mock.calls.length, 0);
Expand All @@ -115,7 +115,11 @@ describe('Ans104Unbundler', () => {
root_tx_id: 'root_tx_id',
} as UnbundleableItem;

await ans104Unbundler.queueItem(mockItem, false, true);
await ans104Unbundler.queueItem({
item: mockItem,
prioritized: false,
bypassBundleFilter: true,
});

assert.deepEqual(
(mockAns104Parser.parseBundle as any).mock.calls[0].arguments[0],
Expand All @@ -124,6 +128,7 @@ describe('Ans104Unbundler', () => {
parentIndex: undefined,
rootParentOffset: 0,
rootTxId: 'root_tx_id',
bypassDataItemFilter: false,
},
);
});
Expand Down
35 changes: 24 additions & 11 deletions src/workers/ans104-unbundler.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,11 @@ export class Ans104Unbundler {
private workerCount: number;
private maxQueueSize: number;
private queue: queueAsPromised<
{ item: UnbundleableItem; bypassFilter: boolean },
{
item: UnbundleableItem;
bypassBundleFilter: boolean;
bypassDataItemFilter: boolean;
},
void
>;
private shouldUnbundle: () => boolean;
Expand Down Expand Up @@ -103,11 +107,17 @@ export class Ans104Unbundler {
this.shouldUnbundle = shouldUnbundle;
}

async queueItem(
item: UnbundleableItem,
prioritized: boolean | undefined,
bypassFilter = false,
): Promise<void> {
async queueItem({
item,
prioritized,
bypassBundleFilter = false,
bypassDataItemFilter = false,
}: {
item: UnbundleableItem;
prioritized?: boolean;
bypassBundleFilter?: boolean;
bypassDataItemFilter?: boolean;
}): Promise<void> {
const log = this.log.child({ method: 'queueItem', id: item.id });

if (this.workerCount === 0) {
Expand All @@ -122,11 +132,11 @@ export class Ans104Unbundler {

if (prioritized === true) {
log.debug('Queueing prioritized bundle...');
this.queue.unshift({ item, bypassFilter });
this.queue.unshift({ item, bypassBundleFilter, bypassDataItemFilter });
log.debug('Prioritized bundle queued.');
} else if (this.queue.length() < this.maxQueueSize) {
log.debug('Queueing bundle...');
this.queue.push({ item, bypassFilter });
this.queue.push({ item, bypassBundleFilter, bypassDataItemFilter });
log.debug('Bundle queued.');
} else {
log.debug('Skipping unbundle, queue is full.');
Expand All @@ -135,10 +145,12 @@ export class Ans104Unbundler {

async unbundle({
item,
bypassFilter,
bypassBundleFilter,
bypassDataItemFilter,
}: {
item: UnbundleableItem;
bypassFilter: boolean;
bypassBundleFilter: boolean;
bypassDataItemFilter: boolean;
}): Promise<void> {
const log = this.log.child({ method: 'unbundle', id: item.id });
try {
Expand All @@ -153,7 +165,7 @@ export class Ans104Unbundler {
// Data item without root_tx_id (should be impossible)
throw new Error('Missing root_tx_id on data item.');
}
if (bypassFilter || (await this.filter.match(item))) {
if (bypassBundleFilter || (await this.filter.match(item))) {
log.info('Unbundling bundle...');
let rootParentOffset = 0;

Expand All @@ -171,6 +183,7 @@ export class Ans104Unbundler {
parentId: item.id,
parentIndex: item.index,
rootParentOffset,
bypassDataItemFilter,
});
log.info('Bundle unbundled.');
}
Expand Down
Loading