Skip to content

hotfix(api): changed how old/new subscriptions are compared #434

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

Merged
thesophiaxu merged 1 commit into from
Apr 4, 2022
Merged

hotfix(api): changed how old/new subscriptions are compared #434

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
4 changes: 2 additions & 2 deletions packages/unigraph-dev-backend/src/dgraphClient.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,13 +274,13 @@ export default class DgraphClient {
* @param query
* @param vars
*/
async queryDgraph(query: string, vars: Record<string, any>|undefined = undefined): Promise<any[]> {
async queryDgraph(query: string, vars: Record<string, any>|undefined = undefined, withTxn?: boolean): Promise<any[]> {
const res = await this.dgraphClient
.newTxn({ readOnly: true })
.queryWithVars(query, vars).catch(e => {console.log(e); return e});
const tns = res.getLatency().getTotalNs();
if (tns > 400000000) console.log(`[PERF] Slow - Transaction complete - but took ${tns / 1000000.0} ms. ` + query.slice(0, 100) + '...')
return Object.values(res.getJson());
return withTxn ? [(res as dgraph.Response).getTxn()?.getStartTs(), Object.values(res.getJson())] : Object.values(res.getJson());
}

/**
Expand Down
3 changes: 2 additions & 1 deletion packages/unigraph-dev-backend/src/server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ export default async function startServer(client: DgraphClient) {
await checkOrCreateDefaultDataModel(client);

// Initialize subscriptions
const pollCallback: MsgCallbackFn = (newdata, sub, ofUpdate, supplementary?) => {
const pollCallback: MsgCallbackFn = (newdata, sub, ofUpdate, supplementary?, txn?) => {
if (sub?.callbackType === 'messageid') {
const msgPort = sub.msgPort!;
if (msgPort?.readyState === 1) {
Expand All @@ -121,6 +121,7 @@ export default async function startServer(client: DgraphClient) {
type: 'subscription',
updated: true,
id: sub.id,
txn,
result: newdata,
ofUpdate,
supplementary,
Expand Down
30 changes: 18 additions & 12 deletions packages/unigraph-dev-backend/src/subscriptions.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,12 +90,18 @@ export function buildPollingQuery(subs: { id: any; query: any }[], states: any)
}, '{')} }`;
}

export type MsgCallbackFn = (updated: any, sub: Subscription, ofUpdate?: number | string, supplementary?: any) => any;
export type MsgCallbackFn = (
updated: any,
sub: Subscription,
ofUpdate?: number | string,
supplementary?: any,
txn?: any,
) => any;

export type MergedSubscription = {
subscriptions: Subscription[];
aggregateQuery: Query;
resolver: (updated: any, ofUpdate?: any) => void;
resolver: (updated: any, ofUpdate?: any, txn?: string) => void;
};

export function mergeSubscriptions(
Expand All @@ -104,13 +110,13 @@ export function mergeSubscriptions(
ids?: any[],
states?: any,
): MergedSubscription[] {
function callbackIfChanged(updated: any, sub: Subscription, ofUpdate: any, supplementary?: any) {
function callbackIfChanged(updated: any, sub: Subscription, ofUpdate: any, supplementary?: any, txn?: string) {
if (
stringify(updated, { replacer: getCircularReplacer() }) !==
stringify(sub.data, { replacer: getCircularReplacer() })
) {
sub.data = updated;
msgCallback(updated, sub, ofUpdate, supplementary);
msgCallback(updated, sub, ofUpdate, supplementary, txn);
}
}

Expand Down Expand Up @@ -138,9 +144,9 @@ export function mergeSubscriptions(
totalMerged.push({
subscriptions: subs,
aggregateQuery: query,
resolver: (updated: any, ofUpdate: any) => {
resolver: (updated: any, ofUpdate: any, txn?: string) => {
subs.forEach((el) => {
callbackIfChanged(updated, el, ofUpdate);
callbackIfChanged(updated, el, ofUpdate, undefined, txn);
});
},
});
Expand All @@ -162,9 +168,9 @@ export function mergeSubscriptions(
totalMerged.push({
subscriptions: subs,
aggregateQuery: query,
resolver: (updated: any, ofUpdate: any) => {
resolver: (updated: any, ofUpdate: any, txn?: string) => {
subs.forEach((el) => {
callbackIfChanged(updated, el, ofUpdate);
callbackIfChanged(updated, el, ofUpdate, undefined, txn);
});
},
});
Expand Down Expand Up @@ -203,7 +209,7 @@ export function mergeSubscriptions(
totalMerged.push({
subscriptions: subs,
aggregateQuery: query,
resolver: (updated: any, ofUpdate: any) => {
resolver: (updated: any, ofUpdate: any, txn?: string) => {
subs.forEach((el) => {
const uidResolver = (uu: string) => (uu.startsWith('$/') ? states.namespaceMap[uu].uid : uu);
const allUids = (el.query as QueryObject).uid;
Expand All @@ -216,7 +222,7 @@ export function mergeSubscriptions(
(Array.isArray((el.query as QueryObject).uid) ? (el.query as QueryObject).uid.length : 1)
)
return;
callbackIfChanged(updatedIts, el, ofUpdate, updated);
callbackIfChanged(updatedIts, el, ofUpdate, updated, txn);
});
},
});
Expand Down Expand Up @@ -271,9 +277,9 @@ export async function pollSubscriptions(
} else query = buildPollingQuery([{ query: el.aggregateQuery, id: getRandomId() }], serverStates);
try {
// const startTime = new Date().getTime();
const results: any[] = await client.queryDgraph(query);
const [txn, results]: any[] = await client.queryDgraph(query, undefined, true);
const val = results[0];
el.resolver(val, ofUpdate);
el.resolver(val, ofUpdate, txn);
// el.queryTime = new Date().getTime() - startTime;
} catch (e) {
console.log(e, query);
Expand Down
11 changes: 4 additions & 7 deletions packages/unigraph-dev-common/src/api/unigraph.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ export default function unigraph(url: string, browserId: string): Unigraph<WebSo
const subscriptions: Record<string, Function> = {};
const subResults: Record<string, any> = {};
const subFakeUpdates: Record<string, any[]> = {};
const subBlockingRoutine: Record<string, any> = {};
const subsTxn: Record<string, any> = {};
const states: Record<string, AppState> = {};
const caches: Record<string, any> = {
namespaceMap: isJsonString(window.localStorage.getItem('caches/namespaceMap'))
Expand Down Expand Up @@ -204,7 +204,7 @@ export default function unigraph(url: string, browserId: string): Unigraph<WebSo
cacheCallbacks[parsed.name]?.forEach((el) => el(parsed.result));
}
if (parsed.type === 'subscription' && parsed.id && subscriptions[parsed.id] && parsed.result) {
if (subBlockingRoutine[parsed.id] && !parsed.ofUpdate) {
if (subsTxn[parsed.id] && parsed.txn < subsTxn[parsed.id]) {
return ev;
}
if (
Expand All @@ -214,11 +214,8 @@ export default function unigraph(url: string, browserId: string): Unigraph<WebSo
) {
return ev;
}
// Reconciled subscription with client, blocking polling/routine updates for 5 seconds
subBlockingRoutine[parsed.id] = true;
setTimeout(() => {
subBlockingRoutine[parsed.id] = false;
}, 5000);

subsTxn[parsed.id] = parsed.txn || 999999999999;

// Now we can safely update the state
subFakeUpdates[parsed.id] = [];
Expand Down
2 changes: 1 addition & 1 deletion packages/unigraph-dev-explorer/src/components/ObjectView/DynamicObjectListView.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,7 +321,7 @@ function DynamicList({

React.useEffect(() => {
setupProps?.onUpdate(items.map((el: any) => itemGetter(el).uid));
}, [items.map((el: any) => itemGetter(el).uid)]);
}, [JSON.stringify(items.map((el: any) => itemGetter(el).uid).sort())]);

return (
<InfiniteScroll
Expand Down