// NOTE: This should be imported at the top of all entrypoints. BUT, if it is missed, // this should ensure it is at least included once (even though including it after // a file prevents it from transforming it, so we'll miss a lot of files doing it this late). import "../inject"; // Shim Promise.race, as no one wants their promises to leak... import { PromiseRace } from "socket-function/src/promiseRace"; Promise.race = PromiseRace; import { shimDateNow } from "socket-function/time/trueTimeShim"; shimDateNow(); import { isNode, isNodeTrue, nextId, timeInMinute, timeInSecond, timeoutToUndefined } from "socket-function/src/misc"; import { SocketFunction } from "socket-function/SocketFunction"; import { isHotReloading, watchFilesAndTriggerHotReloading } from "socket-function/hot/HotReloadController"; import { RequireController, setRequireBootRequire } from "socket-function/require/RequireController"; import { cache, cacheLimited, lazy } from "socket-function/src/caching"; import { getIdentityCAPromise, getOwnMachineId, getOwnThreadId, getThreadKeyCert, verifyMachineIdForPublicKey } from "sliftutils/misc/https/certs"; import { getHostedIP, getSNICerts, publishMachineARecords } from "../-e-certs/EdgeCertController"; import { debugCoreMode, registerGetCompressNetwork, authorityStorage, enableDebugRejections, getNextTime } from "../0-path-value-core/pathValueCore"; import { clientWatcher, ClientWatcher } from "../1-path-client/pathValueClientWatcher"; import { SyncWatcher, proxyWatcher, specialObjectWriteValue, isSynced, PathValueProxyWatcher, atomic, doAtomicWrites, noAtomicSchema, undeleteFromLookup, registerSchemaPrefix, WatcherOptions, doProxyOptions } from "../2-proxy/PathValueProxyWatcher"; import { isInProxyDatabase, rawSchema } from "../2-proxy/pathDatabaseProxyBase"; import { isValueProxy2, getProxyPath } from "../2-proxy/pathValueProxy"; import { getCurrentCallAllowUndefined, getCurrentCall, CallSpec, PathFunctionRunner } from "../3-path-functions/PathFunctionRunner"; import { logErrors } from "../errors"; import { getLastPathPart, getPathIndexAssert, getPathStr2 } from "../path"; import { PermissionsCheck } from "./permissions"; import { inlineNestedCalls, syncSchema } from "../3-path-functions/syncSchema"; import type { identityStorageKey, IdentityStorageType } from "sliftutils/misc/https/certs"; import { ExternalRenderClass, qreact } from "../4-dom/qreact"; import { configRootDiscoveryLocation, getOwnNodeId, onNodeBroadcasted, onNodeDiscoveryReady } from "../-f-node-discovery/NodeDiscovery"; import { pathValueCommitter } from "../0-path-value-core/PathValueCommitter"; import debugbreak from "debugbreak"; import { registerDynamicResource } from "../diagnostics/trackResources"; import { registerPreshutdownHandler, shutdown } from "../diagnostics/periodic"; import { sha256 } from "js-sha256"; import { minify_sync } from "terser"; import { isClient } from "../config2"; import { waitForFirstTimeSync } from "socket-function/time/trueTimeShim"; import { logMeasureTable, measureBlock, measureFnc, measureWrap, startMeasure } from "socket-function/src/profiling/measure"; import { delay } from "socket-function/src/batching"; import { MaybePromise } from "socket-function/src/types"; import { devDebugbreak, getDomain, isBootstrapOnly, isDynamicallyLoading, isPublic, isRecovery, noSyncing } from "../config"; import { Schema2, Schema2T, t } from "../2-proxy/schema2"; import { CALL_PERMISSIONS_KEY } from "./permissionsShared"; import yargs, { check } from "yargs"; import { parseArgsFactory } from "../misc/rawParams"; import * as typesafecss from "typesafecss"; import "../library-components/urlResetGroups"; import { createLocalSchema } from "./schemaHelpers"; typesafecss.setMeasureBlock(measureBlock); typesafecss.setDelayFnc(async fnc => { await (clientWatcher.waitForBeforeTriggerFinished() || Promise.resolve()); fnc(); }); export { t }; let yargObj = parseArgsFactory() .option("verboseall", { type: "boolean", desc: "Log all everything (except network)" }) .option("verbose", { type: "boolean", desc: "Log all writes and reads" }) .option("verbosewrites", { type: "boolean", desc: "Log all writes" }) .option("verbosereads", { type: "boolean", desc: "Log all reads" }) .option("verbosecalls", { type: "boolean", desc: "Log all calls" }) .option("verbosenetwork", { type: "boolean", desc: "Log all network activity" }) .option("verboseframework", { type: "boolean", desc: "Log internal SocketFunction framework" }) .option("nodelay", { type: "boolean", desc: "Don't delay committing functions, even ones that are marked to be delayed." }) .option("hot", { type: "boolean", desc: "force hot reloading to turn on even if public is true." }) .option("ngc", { type: "boolean", desc: "Never GC values, useful for debugging." }) .argv ; setImmediate(() => { if (yargObj.verbose) { ClientWatcher.DEBUG_WRITES = true; ClientWatcher.DEBUG_READS = true; } if (yargObj.verbosewrites) { ClientWatcher.DEBUG_WRITES = true; } if (yargObj.verbosereads) { ClientWatcher.DEBUG_READS = true; } if (yargObj.verbosecalls) { PathFunctionRunner.DEBUG_CALLS = true; Querysub.DEBUG_CALLS = true; } if (yargObj.verboseall) { ClientWatcher.DEBUG_WRITES = true; ClientWatcher.DEBUG_READS = true; PathFunctionRunner.DEBUG_CALLS = true; Querysub.DEBUG_CALLS = true; authorityStorage.DEBUG_UNWATCH = true; enableDebugRejections(); } if (yargObj.verbosenetwork) { SocketFunction.logMessages = true; } if (yargObj.verboseframework) { SocketFunction.silent = false; } if (yargObj.nodelay) { Querysub.DELAY_COMMIT_DELAY = 0; } if (yargObj.ngc) { authorityStorage.DEBUG_NEVER_GC = true; } }); if (!registerGetCompressNetwork) { devDebugbreak(); } export type MachineSourceCheck = { /** NOTE: The source for this is added inline, and so it cannot use any external variables. */ getExtraDataClientsideInline?: () => T | "nosource"; shouldMachineIdSeeSource: (config: { machineId: string, extra: T }) => Promise; } export function id(obj: unknown) { let path = isValueProxy2(obj); if (!path) { return ""; } return getLastPathPart(path); } export class Querysub { public static syncSchema = syncSchema; public static qreact = qreact; public static Component = qreact.Component; public static render = qreact.render; public static context = qreact.context; public static createElement = qreact.createElement; public static Fragment = qreact.Fragment; public static t = t; public static shutdown = () => shutdown(); public static onShutdown = (fnc: () => Promise) => registerPreshutdownHandler(fnc); /** IMPORTANT! New schemas must be deployed by calling `yarn deploy` const { data } = syncSchema<{ width: number; height: number; }>()({ // If domainName is LOCAL_DOMAIN, the values are kept in memory in the current process. domainName: getDomain(), functions: { setWidth, }, module, moduleId: "sharedParams", }); function setWidth(width: number) { data().width = width; } */ public static createSchema = syncSchema; public static noAtomicSchema = (code: () => T) => noAtomicSchema(code); /** Undelete is a bit dangerous, and races with garbage collection. Using gcDelay * helps reduce this race, but it is still possible to undelete, and then * have the value disappear after a few minutes. * - The race occurs if an undelete happens less than ~5 * MAX_CHANGE_AGE * before we start to GC. * - This race is more of a propagation delay. The value is already past the point * of no return, and so the undeletion cannot work. IF you set a large enough gcDelay * (a month), then not only is this unlikely to happen, but it means it will only happen * if the user waits a month to undelete it. Because undeletions aren't random, and are * more likely to occur shorter after the deletion, this means the chance is really less * than 5 minutes / 1 month (< 1/100K). */ public static undeleteFromLookup = undeleteFromLookup; public static WILDCARD = "*"; public static WILDCARD_NUM = "*" as any as number; public static createLocalSchema = createLocalSchema; public static flushDelayedFunctions = () => flushDelayedFunctions(); public static DEBUG_CALLS = false; public static DEBUG_PREDICTIONS = false; public static PREDICT_CALLS = true; // These audits are mostly wrong, so I'm just disabling them. public static AUDIT_PREDICTIONS = false; public static SIMULATE_LAG = 0; public static registerAliveChecker = (config: AliveChecker) => registerAliveChecker(config); public static registerGarbageCollection = (config: AliveChecker) => registerAliveChecker(config); /** Delay used when functions are specified as delayCommit */ public static DELAY_COMMIT_DELAY = 1000 * 3; /** Time predictions last for, after which point we remove them irregardless of if we have received a response. * - NOTE: If this is too short, predictions stop before the server can finish. If it's too long, * server issues (such as the FunctionRunner being missing) won't be noticed. So... * we SHOULD increase this when our setup is more stable (60 seconds is probably good). * SHOULD be > DELAY_COMMIT_DELAY, Otherwise delayed commits are going to cause predictions to be rejected before we even submit the actual call. */ public static PREDICTION_MAX_LIFESPAN = 1000 * 15; /** Fairly lax, to handle lag during initial connection setup. * AND THEN, VERY lax, to handle server lag. */ public static MAX_FUTURE_CALL_TIME = 10_000; public static COMPRESS_NETWORK = true; public static now = getSyncedTime; public static time = getSyncedTime; public static timeUnique = getSyncedTimeUnique; public static getCallTime = getSyncedTime; public static getFunctionCallTime = getSyncedTime; // NOTE: This is just an ID inside of a call that will be unique. It's not the call ID that the function uses when it's running. That's something that's different, and that's defined in PathFunctionHelpers public static getCallId = () => Querysub.getCallerMachineId() + "_" + Querysub.getCallTime(); // Returns a random value between 0 and 1. The value depends on the callId, and index (being identical for // the same callid and index). public static callRandom = () => hashRandom(Querysub.getCallId(), getNextCallIndex()); public static nextId = () => { if (!Querysub.isInSyncedCall()) return nextId(); return Querysub.getCallId() + "_" + getNextCallIndex(); }; public static getNextCallIndex = getNextCallIndex; public static configRootDiscoveryLocation = configRootDiscoveryLocation; public static nowDelayed = timeDelayed; public static timeDelayed = timeDelayed; public static getCallerMachineId() { let call = getCurrentCallAllowUndefined(); if (!call) return getOwnMachineId(getDomain()); return getCurrentCall().callerMachineId; } public static getCallerIP = () => getCurrentCall().callerIP; public static getCallerIPAllowUndefined = () => getCurrentCallAllowUndefined()?.callerIP; public static CALL_PERMISSIONS_KEY = CALL_PERMISSIONS_KEY; public static watchCurrentFnc = () => PathValueProxyWatcher.BREAK_ON_CALL.add( getPathStr2(getCurrentCall().ModuleId, getCurrentCall().FunctionId) ); public static unwatchCurrentFnc = () => PathValueProxyWatcher.BREAK_ON_CALL.delete( getPathStr2(getCurrentCall().ModuleId, getCurrentCall().FunctionId) ); public static watchWrites = (value: unknown) => proxyWatcher.debug_breakOnWrite(value); public static debugWrites = (value: unknown) => proxyWatcher.debug_breakOnWrite(value); public static breakOnWrite = (value: unknown) => proxyWatcher.debug_breakOnWrite(value); public static logWrites = (value: unknown) => proxyWatcher.debug_logOnWrite(value); public static reuseLastWatches = () => proxyWatcher.reuseLastWatches(); /** Gets the path, adding a watch to the path as well. */ public static getPath = (value: unknown) => proxyWatcher.getPathAndWatch(value); public static getCountPerPaint = (type: string) => getCountPerPaint(type); /** NOTE: While breakOnNextTrigger is correct, is tends to break too late. It is easier * if you use breakOnNextTrigger2. */ public static breakOnNextTrigger = () => clientWatcher.debugBreakOnNextTrigger(); // More of a hack, as it guesses the paths will the same, but is invaluable when it works. public static breakOnNextTrigger2 = () => { let triggerReason = Querysub.getTriggerReason(); for (let path of triggerReason?.paths || []) { Querysub.breakOnWrite(path); } }; public static watchUnsyncedComponents = () => qreact.watchUnsyncedComponents(); public static watchAnyUnsyncedComponents = () => qreact.watchUnsyncedComponents().size > 0; public static cancelRender = () => qreact.cancelRender(); /** Calls the callback any time components are mounted to do the DOM (which happens after * time they render, not just the first time). */ public static globalAfterRenderWatch = (callback: (components: ExternalRenderClass[]) => void) => qreact.globalAfterRenderWatch(callback); public static doAtomicWrites = (callback: () => T): T => doAtomicWrites(callback); public static trustedDomains = new Set(); /** Wait for some startup initialization. Not necessary, but can ensure any initial calls * are in a better state. * - Useful for client service scripts. * NOTE: waitForFirstTimeSync is called in Socket.mount, so this function * isn't presently necessary in most serverside scripts. */ @measureFnc public static async optionalStartupWait(addTime?: (name: string) => void) { // Used by authorityLookup, pulled out here so timing is more clear. await onNodeDiscoveryReady(); addTime?.("onNodeDiscoveryReady"); await authorityLookup.startSyncing(); addTime?.("authorityLookup.startSyncing"); await measureBlock(async () => { await waitForFirstTimeSync(); }, "waitForFirstTimeSync"); addTime?.("waitForFirstTimeSync"); } public static createWatcher(watcher: (obj: SyncWatcher) => void, options?: Partial>): { dispose: () => void; explicitlyTrigger: () => void; } & SyncWatcher { return proxyWatcher.createWatcher({ baseFunction: watcher, canWrite: true, ...options, watchFunction: () => { return watcher(proxyWatcher.getTriggeredWatcher()); }, }); } public static createWriteWatcher(watcher: (obj: SyncWatcher) => void) { return this.createWatcher(watcher); } public static undoDelete(holder: { [key: string]: unknown }, key: string) { holder[key] = specialObjectWriteValue; } public static localWriteWrapper(fnc: (...args: Args[]) => void): (...args: Args[]) => void { return (...args: Args[]) => Querysub.serviceWriteDetached(() => fnc(...args), { debugName: fnc.name, }); } public static runSynced = Querysub.serviceWriteDetached; public static commit = Querysub.serviceWriteDetached; public static write = Querysub.serviceWriteDetached; // NOTE: We might want to add a kind of commit function that has finish and start order set by default, as this is kind of what most people assume. However, it adds so much lag that it definitely shouldn't be the default, even when we're client-side. public static serviceWriteDetached(fnc: () => unknown, options?: Partial>) { logErrors(proxyWatcher.commitFunction({ canWrite: true, watchFunction: fnc, //finishInStartOrder: isClient(), noWaitForCommit: isClient(), ...options, })); } public static commitDelayed = (fnc: () => unknown) => { setImmediate(() => { Querysub.serviceWriteDetached(fnc); }); }; public static exit = Querysub.gracefulTerminateProcess; public static async gracefulTerminateProcess() { await pathValueCommitter.waitForValuesToCommit(); if (isNode()) { process.exit(); } }; public static syncedCommit = Querysub.serviceWrite; public static commitSynced = Querysub.serviceWrite; public static commitAsync = Querysub.serviceWrite; public static async serviceWrite(fnc: () => T, options?: Partial>) { let calls: CallSpec[] = []; let result = await proxyWatcher.commitFunction({ canWrite: true, watchFunction: fnc, onCallCommit(call, metadata) { calls.push(call); }, // The only reason we really wait for the commit is for server side so we don't terminate the process before the write finishes. But if we are client side this isn't an issue. noWaitForCommit: isClient(), ...options, }); // Wait for calls to predict. This SHOULDN'T take very long. This allows us to use serviceWrite // in async contexts, having function calls locally predict so the next call can use the value. for (let call of calls) { await onCallPredict(call); } return result; } public static localCommit = Querysub.localRead; public static commitLocal = Querysub.localRead; public static fastRead(fnc: () => T, options?: Partial>) { return proxyWatcher.runOnce({ watchFunction: fnc, ...options, allowProxyResults: true, }); } public static fastReadAsync(fnc: () => T, options?: Partial>) { return Querysub.serviceWrite(fnc, { ...options, allowProxyResults: true, noWaitForCommit: true }); } public static readLocal = Querysub.localRead; public static localRead(fnc: () => T, options?: Partial>) { return proxyWatcher.runOnce({ watchFunction: fnc, ...options, }); } public static async unsafeInlineFunctionsServiceWrite(code: () => T) { return await Querysub.serviceWrite(() => { return inlineNestedCalls(code); }); } public static undefinedIfAnyUnsynced(get: () => T) { let count = { value: 0 }; let result = proxyWatcher.countUnsynced(count, get); if (count.value > 0) return undefined; return result; } public static ignorePermissionsChecks = (code: () => T) => PermissionsCheck.skipPermissionsChecks(code); public static skipPermissionsChecks = (code: () => T) => PermissionsCheck.skipPermissionsChecks(code); public static anyUnsynced() { return !Querysub.allSynced(); } public static allSynced(config?: { ignoreAlwaysCommitAllRunsFlag?: boolean }) { return proxyWatcher.isAllSynced(config); } public static fullySynced = Querysub.allSynced; public static isFullySynced = Querysub.allSynced; public static isAllSynced = Querysub.allSynced; public static isAnyUnsynced = Querysub.anyUnsynced; public static isInWatcher = () => !!proxyWatcher.getTriggeredWatcherMaybeUndefined(); public static onNextDisconnect = SocketFunction.onNextDisconnect; public static isNodeConnected = SocketFunction.isNodeConnected; public static pathHasAnyWatchers(get: () => unknown) { return clientWatcher.pathHasAnyWatchers(getProxyPath(get)); } public static onDispose = (callback: () => void) => qreact.onDispose(callback); public static runAsync = Querysub.onCommitFinished; public static onCommitFinished(callback: () => void) { if (!proxyWatcher.getTriggeredWatcherMaybeUndefined()) { void Promise.resolve().finally(callback); } else if (proxyWatcher.getTriggeredWatcher().options.temporary) { let innerDisposed = proxyWatcher.getTriggeredWatcher().onInnerDisposed; if (innerDisposed.includes(callback)) return; innerDisposed.push(callback); } else { let afterTriggered = proxyWatcher.getTriggeredWatcher().onAfterTriggered; if (afterTriggered.includes(callback)) return; afterTriggered.push(callback); } } public static onCommitFinishedCommit(callback: () => void) { this.onCommitFinished(() => { Querysub.commit(callback); }); } public static onSynced(callback: () => void) { let watcher = proxyWatcher.getTriggeredWatcherMaybeUndefined(); Querysub.onCommitFinished(() => { if (!watcher || !watcher.hasAnyUnsyncedAccesses()) { callback(); } }); } /** Checks all recursive watches. However... if something writes to a value you watch (ex, DerivedCache), we can't track that, so this is of limited use. */ public static onNestedSynced(callback: () => void) { onNestedSynced(callback); } public static noLocks = (callback: () => void) => { // I don't think there's really any reason to use the regular no-locks, as it's only for local paths, but... I think we don't use locks for local paths anyway? doProxyOptions({ unsafeNoLocks: true }, callback); }; /** A more powerful version of omCommitFinished, which even waits for call predictions (or tries to). * - Also see afterPredictionsSynced, which runs the callback in a write. * NOTE: IDEALLY, you would just call a series of synced functions. They will be run in order, * with all the synced data, and just work. However... if you are running very expensive * asynchronous operations, you might need to use an async workflow. * "afterPredictions" helps in async workflows, by calling your callback after all local function * predictions finish. This allows you to (somewhat) read back their writes. It isn't perfect, * but it is safer then Promise.resolve().then(callback), or any type of setTimeout. Usually * the callback will be fast, except for the first time, when code will needed by loaded to * run the predictive functions. * NOTE: Only waits for calls BEFORE the callback is called. This is useful, but also means * you can't call this at the start of your function (as nothing will have been called yet). * NOTE: This can also be used to prevent */ public static afterPredictions(callback: () => MaybePromise) { let calls = proxyWatcher.getTriggeredWatcher().pendingCalls.map(x => x.call); Querysub.onCommitFinished(async () => { await Promise.all(calls.map(x => Querysub.onCallPredict(x))); logErrors(callback()); }); } public static afterAllRendersFinished(callback: () => void) { // onCommitFinished prevents duplicates, as well as only running when we are actually done Querysub.onCommitFinished(async () => { await clientWatcher.waitForTriggerFinished(); // No idea why this work, but... we do need to wait for promises... twice. Maybe we are waiting for some promise stacks to unwind? But twice? Bizarre // - Needed for ReactEditor:updateSelectionRects AND LazyRenderList... await Promise.resolve(); await Promise.resolve(); callback(); }); } public static async afterAllRendersFinishedPromise() { return new Promise(r => { Querysub.afterAllRendersFinished(r); }); } /** Solely for use to prevent local writes from occuring before predictions. We MIGHT make this a framework thing, * or just not bother with it (as after a prediction is run once the code will be loaded allowing it to always * run before the next frame). */ public static afterPredictionsSynced(callback: () => void) { Querysub.afterPredictions(async () => { Querysub.serviceWriteDetached(callback); }); } /** Returns true if any predictions are running. In which case, we will correctly rerun the function and consider it non-synced until the predictions finish. This allows you to check for predictions of the in your function and return if you have any. This is useful for functions that need a stable state before they run. However, if many functions use this and you trigger them at once, it might result in an n-squared situation, so this should be used with caution. * - The best use case is if something is triggering maybe on focus and on blur, and you want to make sure the on blur changes happen before your on focus changes happen. */ public static syncAnyPredictionsPending(): boolean { let waitPromise = onAllPredictionsFinished(); if (waitPromise) { proxyWatcher.triggerOnPromiseFinish(waitPromise, { waitReason: "Waiting for predictions to finish", }); return true; } return false; } public static onCallPredict = (x: CallSpec | undefined) => onCallPredict(x); /** NOTE: USUALLY you don't have to wait for predictions, as functions will run in order anyways. * BUT, if you run a synced function in the UI, then somewhere unrelated try to read values * for offload writing (which isn't ideal, but sometimes is necessary), then waiting * for predictions to finish can help in the rare cases where predictions are slow * (such as if code is still loading). */ public static waitUntilAllPredictionsFinished = () => waitUntilAllPredictionsFinish(); /** Useful to prevent in-progress state from showing temporarily. */ public static waitForPredictionsSynced = () => { if (anyPredictionsPending()) { proxyWatcher.triggerOnPromiseFinish(Querysub.waitUntilAllPredictionsFinished(), { waitReason: "Waiting for predictions to finish", }); } }; public static onCommitPredictFinished = this.onCallPredict; public static getOwnMachineId = () => getOwnMachineId(getDomain()); public static getSelfMachineId = () => getOwnMachineId(getDomain()); public static getOwnNodeId = () => getOwnNodeId(); public static getSelfNodeId = () => getOwnNodeId(); public static getOwnThreadId = () => getOwnThreadId(getDomain()); /** Set ClientWatcher.DEBUG_SOURCES to true for to be populated */ public static getTriggerReason() { return proxyWatcher.getTriggeredWatcher().triggeredByChanges; } public static isInSyncedCall() { return proxyWatcher.inWatcher() && isInProxyDatabase(); } public static isSynced(value: unknown | (() => unknown)) { let path = Querysub.getPath(value); if (!path) return true; return authorityStorage.isSynced(path); } public static ignoreWatches(code: () => T) { return proxyWatcher.ignoreWatches(code); } public static assertDomainAllowed(path: string) { let domain = getPathIndexAssert(path, 0); if (!Querysub.trustedDomains.has(domain)) { throw new Error(`Domain ${domain} is not trusted on this server`); } } private static pendingLazyCommit: undefined | { fncs: (() => unknown)[]; }; private static getLazyCommit(delayTime: number = 100) { const TARGET_FRACTION_TIME = 0.1; if (this.pendingLazyCommit) { return this.pendingLazyCommit; } this.pendingLazyCommit = { fncs: [], }; let obj = this.pendingLazyCommit; setTimeout(async () => { let fncs = obj.fncs; obj.fncs = []; // Go back to non-active mode if there are no triggers if (fncs.length === 0) { this.pendingLazyCommit = undefined; return; } let startTime = Date.now(); //console.log(`Batched ${fncs.length} functions in lazy commit`); await Querysub.commitSynced(() => { for (let fnc of fncs) { try { fnc(); } catch (e) { console.error(`Error in lazy commit`, e); } } }); await clientWatcher.waitForTriggerFinished(); //await delay("afterPaint"); let totalTimeCaused = Date.now() - startTime; totalTimeCaused = Math.max(16, totalTimeCaused); let delayTime = totalTimeCaused / TARGET_FRACTION_TIME; delayTime = Math.min(5000, delayTime); //console.log(`Lazy commit took ${totalTimeCaused}ms, setting delay time to ${delayTime}ms`); let queuedWhileCommitting = obj.fncs; this.pendingLazyCommit = undefined; // Create a new commit with our new delay time let newCommit = this.getLazyCommit(delayTime); newCommit.fncs.push(...queuedWhileCommitting); }, delayTime); return obj; } /** Waits to commit * - Might reorder other types of commit calls, but WILL NOT reorder commitLazy calls * - Batches fncs into single functions (which might result in extra dependencies / * extra rejections of unrelated functions). * - Tries to track triggered time from these commits, and keep it below 10%, * batching more and more to make sure this is the case. */ public static commitLazy( fnc: () => unknown, ) { this.getLazyCommit().fncs.push(fnc); } private static synchronousStartTime = 0; /** To be called in a loop, frequently, and when enough time has passed without a paint, * it will return a promise which delays long enough until we can paint again. * - Works in conjunctions with other yieldForPaint calls, so if many loops are * running we won't unnecessarily wait in all of them. * @param yieldAfterTime, if there are multiple concurrent callers the first one will * determine the used time, and both will use that time. */ public static yieldForPaint( yieldAfterTime = 1000 ): MaybePromise { // NOTE: We even wait serverside, as the server doesn't want to get locked up // on synchronous processing either! // NOTE: Painting can easily take 32ms, and even more if we are rendering, // so... we can't wait too frequently, otherwise slow processing scripts (the // only reason to use this function) will be significantly slower. if (!this.synchronousStartTime) { this.synchronousStartTime = Date.now(); void delay("paintLoop").finally(() => { this.synchronousStartTime = 0; }); } let timeSinceStart = Date.now() - this.synchronousStartTime; if (timeSinceStart > yieldAfterTime) { console.log(`Yielded for paint after ${timeSinceStart}ms`); return delay("paintLoop"); } } public static debugMax() { this.debugFramework(); ClientWatcher.DEBUG_READS = true; SocketFunction.silent = false; PathValueProxyWatcher.TRACE = true; debugCoreMode(); } public static debugFramework() { console.log(`We are ${Querysub.getSelfMachineId()}`); ClientWatcher.DEBUG_WRITES = true; Querysub.DEBUG_CALLS = true; } public static async debugQreact() { await qreact.debug(); } public static enableSlowPaintTracking(threshold = 100) { if (isNode()) return; if ((globalThis as any).measuringPaints) return; (globalThis as any).measuringPaints = true; void (async function fnc() { while (true) { let paintTime = performance.now(); let measureObj = startMeasure(); await new Promise(r => requestAnimationFrame(r)); paintTime = performance.now() - paintTime; // ALWAYS finish the profile, or we leak a lot of memory! let profile = measureObj.finish(); let profiledTime = Object.values(profile.entries).reduce((a, b) => a + b.ownTime.sum, 0); if (paintTime > threshold && profiledTime > threshold && Object.keys(profile.entries).length > 0) { console.log(`Slow paint time: ${formatTime(paintTime)} (${formatNumber(1000 / paintTime)}fps), profile is:`); logMeasureTable(profile, { minTimeToLog: 0, maxTableEntries: 100, mergeDepth: 2, //thresholdInTable: 0.001, }); logMeasureTable(profile, { minTimeToLog: 0, maxTableEntries: 100, mergeDepth: 3, //thresholdInTable: 0.001, }); } } })(); } private static socketFunctionInit() { SocketFunction.addGlobalHook(() => authorityLookup.startSyncing()); } private static hostCalled = false; private static afterBootImports: string[] = []; /** NOTE: Imports occur one at a time, after the "clientsideEntryPoint", with each import waiting for the previous to finish. */ public static registerAdditionalRoot(path: string) { if (Querysub.hostCalled) { throw new Error(`All "Querysub.registerAfterBootImport" calls MUST be called before "Querysub.hostServer" (to prevent servers from existing in partially loaded states).`); } Querysub.afterBootImports.push(path); } public static async hostServer(config: { // Domains we will run functions from. If this domains have any malicious code deploy (to the FunctionRunner data schema), // we are vulnerable to running this code when we check permissions for syncing paths. trustedDomains: string[]; // The root of your project, likely where your package.json and .git are. rootPath: string; // Ex, `./src/client.ts` // - Relative to rootPath clientsideEntryPoint: string; hotReload: boolean; /** You can also set `port: 0` if any port is fine. */ port: number; justHTTP?: boolean; debugName?: string; staticRoots?: string[]; sourceCheck?: MachineSourceCheck; }) { await getIdentityCAPromise(getDomain()); console.log(`Hosting server with config: ${JSON.stringify(config)}`); this.socketFunctionInit(); if (isClient()) { throw new Error(`--client processes cannot host a service. Either stop passing --client and keep the process on the network and trusted, or stop calling hostServer and call Querysub.configRootDiscoveryLocation instead.`); } Querysub.hostCalled = true; for (let domain of config.trustedDomains) { Querysub.trustedDomains.add(domain); } // Hot reloading on public servers breaks things when we update (as the git pull triggers a hot reload), so... don't do that. if (config.hotReload && !isPublic() || yargObj.hot) { watchFilesAndTriggerHotReloading(); } //SocketFunction.silent = false; //PermissionsCheck.DEBUG = true; //ClientWatcher.DEBUG_WRITES = true; await this.addSourceMapCheck(config); const fs = await import("fs"); RequireController.addMapGetModules(async (result, args) => { await Promise.all(Object.values(result.modules).map(async (mod) => { if (!require.cache[mod.filename]?.sendFullSource) return; // NOTE: This is used for extractType... which is very useful. And... giving the source // isn't that big of a deal anyways... try { mod.originalSource = await fs.promises.readFile(mod.filename, "utf8"); } catch { } })); return result; }); SocketFunction.expose(RequireController); setRequireBootRequire(config.rootPath); let entryPaths = [config.clientsideEntryPoint, ...Querysub.afterBootImports]; SocketFunction.setDefaultHTTPCall(RequireController, "requireHTML", { // NOTE: This should be cached in our CDN anyways, so it should be fast // to access even if every request hit the server (as it will hit the CDN's, // server, which is usually close to the client). And it only impacts the initial // load, which isn't that important for us. cacheTime: timeInMinute * 5, }); RequireController.injectHTMLBeforeStartup(async () => { const { getEdgeBootstrapScript } = await import("../4-deploy/edgeBootstrap"); const { getEdgeNodeConfigURL } = await import("../4-deploy/edgeNodes"); let edgeBootstrapFile = await getEdgeBootstrapScript({ edgeNodeConfigURL: await getEdgeNodeConfigURL(), }); return ``; }); for (let root of config.staticRoots || []) { RequireController.addStaticRoot(root); } if (!noSyncing() && !isBootstrapOnly()) { SocketFunction.expose(QuerysubController); } let allowHostnames: string[] = []; for (let domain of Querysub.trustedDomains) { allowHostnames.push(domain); } allowHostnames.push("127-0-0-1." + getDomain()); if (isBootstrapOnly() && isPublic()) { if (config.port !== 443) { throw new Error(`--bootstraponly requires you to set port 443. There can only be one bootstrap node per server.`); } } let mountedNodeId = await SocketFunction.mount({ public: isPublic(), port: config.port, autoForwardPort: true, ...await getThreadKeyCert(getDomain()), SNICerts: { ...await getSNICerts({ publicPort: config.port, }), }, allowHostnames, }); let port = getNodeIdLocation(mountedNodeId)?.port; let { ip, ipDomain } = await publishMachineARecords(); if (!isBootstrapOnly()) { const { registerEdgeNode } = await import("../4-deploy/edgeNodes"); await registerEdgeNode({ host: ipDomain + ":" + port, entryPaths, }); } } private static async addSourceMapCheck(config: { sourceCheck?: MachineSourceCheck; }) { const sourceCheck = config.sourceCheck; if (!sourceCheck) return; let { getExtraDataClientsideInline, shouldMachineIdSeeSource, } = sourceCheck; let ed25519 = await import("../-a-auth/ed25519"); const ed25519Module = require.cache[require.resolve("../-a-auth/ed25519")]; if (!ed25519Module) { throw new Error(`Failed to find imported ed25519 module in require.cache`); } let moduleContents = ed25519Module.moduleContents; if (!moduleContents) { throw new Error(`ed25519 module was missing moduleContents`); } moduleContents = moduleContents.replace(/\/\/# sourceMappingURL=.*\n/g, ""); type SignedIdentity = { payload: { machineId: string; publicKeyB64: string; time: number; extra: ReturnType>; }; signatureB64: string; }; const getExtraDataClientsideSource = (getExtraDataClientsideInline || (() => { })).toString(); function signWrapper(moduleContents: string, getExtraDataClientsideSource: string) { let exports = {} as typeof ed25519; let module = { exports }; function require() { throw new Error("require not supported"); } let fnc = `(function ed25519(exports, require, module, __filename, __dirname, importDynamic) {${moduleContents}\n })`; eval(fnc)(exports, require, module, "inlined", "inlined", require); if (!getExtraDataClientsideSource.startsWith("function") && !getExtraDataClientsideSource.split("\n").includes("=>")) { getExtraDataClientsideSource = "function " + getExtraDataClientsideSource; } const getExtraDataClientside = eval(`(${getExtraDataClientsideSource})`) as Exclude; globalThis.remapImportRequestsClientside = globalThis.remapImportRequestsClientside || []; globalThis.remapImportRequestsClientside.push(async (args) => { try { let key: typeof identityStorageKey = "machineCA_14"; let storageValueJSON = localStorage.getItem(key); if (!storageValueJSON) return args; let storageValue = JSON.parse(storageValueJSON) as IdentityStorageType; let machineId = storageValue.domain.split(".").at(-3) || ""; let pem = Buffer.from(storageValue.keyB64, "base64"); let privateKey = exports.extractRawED25519PrivateKey(pem); let publicKey = await exports.extractPublicKey(privateKey); let extra = getExtraDataClientside(); if (extra === "nosource") return args; let payload: SignedIdentity["payload"] = { machineId, time: Date.now(), extra, publicKeyB64: publicKey.toString("base64"), }; let signature = await exports.signWithPEM({ pem, data: payload, }); let signedIdentity: SignedIdentity = { payload, signatureB64: signature.toString("base64"), }; args[2] = args[2] || {}; let config = args[2] as any; config.signedIdentity = signedIdentity; return args; } catch (e) { console.error(`Error signing request, code won't include sourcesmaps`, e); return args; } }); } moduleContents = `(${signWrapper.toString()})(${JSON.stringify(moduleContents)}, ${JSON.stringify(getExtraDataClientsideSource)})`; RequireController.injectHTMLBeforeStartup(` `); async function isAllowedToSeeSource(signedIdentity: SignedIdentity | undefined): Promise { if (!signedIdentity) return false; let { signatureB64, payload } = signedIdentity; let { publicKeyB64, machineId, time, extra } = payload; let publicKey = Buffer.from(publicKeyB64, "base64"); if (!verifyMachineIdForPublicKey({ machineId, publicKey })) return false; // Requests shouldn't take TOO long, and if they do, it is likely due to cache, and permissions should be denied. let expireTime = Date.now() - timeInMinute * 5; if (time < expireTime) return false; return await shouldMachineIdSeeSource({ machineId, extra }); } const stripSourceBase = cacheLimited(100_000, function stripSourceBase(source: string | undefined) { if (!source) return source; const USE_TERSER = true; if (USE_TERSER) { // NOTE: Using terser results in a bundle of 638KB, instead of 815KB (a 20% reduction), // which is just barely good enough to justify using a library. It is a lot slower too, // but as it is cached, it should be fine... // - It takes about 5s to minify our entire code base, which again, is barely good enough. try { return minify_sync( { file: source }, { mangle: { module: true } } ).code; } catch { return source; } } // Strip inline sourcemaps source = source.replace(/\/\/# sourceMappingURL=.*\n/g, ""); // Strip comments let inSingleLineComment = false; let inMultiLineComment = false; let output = ""; for (let i = 0; i < source.length; i++) { let char = source[i]; if (inSingleLineComment) { if (char === "\n") { inSingleLineComment = false; output += "\n"; } continue; } if (inMultiLineComment) { if (char === "*" && source[i + 1] === "/") { inMultiLineComment = false; i++; } continue; } if (char === "/") { // HACK: To prevent needing to handle strings, we only strip comment at the beginning of the line if (source[i + 1] === "/") { // Find line start let lineStart = source.lastIndexOf("\n", i); if (!source.slice(lineStart, i).trim()) { inSingleLineComment = true; i++; continue; } } if (source[i + 1] === "*") { inMultiLineComment = true; i++; continue; } } output += char; } return output; }); function stripSource(module: SerializedModule) { module = { ...module }; module.source = stripSourceBase(module.source); return module; } if (!isRecovery()) { RequireController.addMapGetModules(async (result, args) => { let configObj = args[2] as { signedIdentity: SignedIdentity | undefined } | undefined; if (!await isAllowedToSeeSource(configObj?.signedIdentity)) { await isAllowedToSeeSource(configObj?.signedIdentity); //console.log(red(`Not allowed to see source`)); for (let [key, value] of Object.entries(result.modules)) { result.modules[key] = stripSource(value); } } else { //console.log(green(`Allowed to see source`)); } return result; }); } } public static async hostService(name: string, port = 0) { if (isClient()) { throw new Error(`--client processes cannot host a service. Either stop passing --client and keep the process on the network and trusted, or stop calling hostServer and call Querysub.configRootDiscoveryLocation instead. You MUST provide configRootDiscoveryLocation a valid nodeId. Which means either you do server selection manually, or if you are developing, just point it to "127-0-0-1.${getDomain()}:your local port here"`); } await getIdentityCAPromise(getDomain()); let times: { name: string; duration: number }[] = []; let time = Date.now(); function addTime(name: string) { times.push({ name, duration: Date.now() - time }); time = Date.now(); } this.socketFunctionInit(); let mountPromise = SocketFunction.mount({ public: isPublic(), port, ...await getThreadKeyCert(getDomain()), }); let publishPromise = publishMachineARecords(); await Promise.all([mountPromise, publishPromise]); addTime("mount & publish a records"); await Querysub.optionalStartupWait(addTime); console.log(magenta(`Started hosting service ${name}`) + ` | ${times.map(t => `${blue(t.name)}: ${green(formatTime(t.duration))}`).join(" | ")}`); } /** Syncs keys AND values (as we won't return a key for a value that is undefined). */ public static keys(obj: T, config?: { /** Value between 0 and 1, used with endFraction to try return a subset of keys. * - Tries to only talk to the necessary servers, being highly efficient if the servers are sharded * (otherwise likely does filtering serverside, which is still somewhat efficient). */ startFraction?: number; /** Value between 0 and 1. * BOTH start and end must be provided, otherwise if only one is provided we will ignore it. */ endFraction?: number; }): (keyof T)[] { let { startFraction, endFraction } = config || {}; if ((startFraction === undefined) !== (endFraction === undefined)) { throw new Error(`startFraction and endFraction must both be provided, or both be undefined. If you want to get all keys, don't provide startFraction and endFraction.`); } if (!isNode() && startFraction !== undefined && endFraction !== undefined) { console.warn(`keys() with a range restriction is not supported clientside. It's too complicated for the proxy to handle it because the hashing depends on the authority server. You can synchronize all the keys client side, but you can't synchronize a restriction of the keys.`); } if (startFraction === undefined || endFraction === undefined) { return Object.keys(obj); } let path = isValueProxy2(obj); if (!path) { return Object.keys(obj); } let packedPath = encodeParentFilter({ path, startFraction, endFraction }); return proxyWatcher.getKeys(packedPath); } // TODO: Maybe expose checkPermissions(getValue: () => unknown)? // - It would be easy, if we every need to explicitly check if we have permissions. Although, it seems // like just relying on the automatic checking is better? } let nextGlobalIndex = 1000 * 1000 * 1000 * 100; let nextCallIndex = 1; let triggeredWatcher: unknown; function getNextCallIndex() { let triggered = proxyWatcher.getTriggeredWatcher(); if (!triggered.options.temporary) { return nextGlobalIndex++; } // By using the watcher, commits can have consistent ids. if (triggeredWatcher !== triggered) { triggered.onAfterTriggered.push(() => { nextCallIndex = 1; triggeredWatcher = undefined; }); triggeredWatcher = triggered; } return nextCallIndex++; } const timeData = createLocalSchema<{ time: number; intervals: { [interval: number]: number; }; }>("time"); const initTimeLoop = lazy(() => { function onNextTime() { requestAnimationFrame(onNextTime); // TODO: We should really stop the loop, and start it again when someone is watching // our values. Which we can either do via getSyncedTime, OR, we could technically // get pathValueClientWatcher to tell us when someone is watching it (but using getSyncedTime // is probably better). if (clientWatcher.pathHasAnyWatchers(getProxyPath(() => timeData().time))) { Querysub.serviceWriteDetached(() => { timeData().time = Date.now(); }); } } onNextTime(); }); let initInterval = cache((interval: number) => { setInterval(() => { if (clientWatcher.pathHasAnyWatchers(getProxyPath(() => timeData().intervals[interval]))) { Querysub.serviceWriteDetached(() => { timeData().intervals[interval] = Date.now(); }); } }, interval); }); function getSetTime() { let watcher = proxyWatcher.getTriggeredWatcherMaybeUndefined(); if (watcher?.options.predictMetadata?.delayCommit) { debugger; console.warn(`Accessed the function time inside of a delayed commit call in ${watcher.debugName}. This means the prediction is almost certainly going to be wrong. You should provide Date.now() as an argument to the function instead. `); } let call = getCurrentCallAllowUndefined(); if (call) { return call.runAtTime.time; } if (watcher?.options.temporary) { return watcher.createTime; } return undefined; } let lastTime = 0; function getSyncedTimeUnique() { let setTime = getSetTime(); if (setTime !== undefined) { return addEpsilons(setTime, getNextCallIndex()); } let time = getSyncedTime(); if (time <= lastTime) { time = addEpsilons(lastTime, 1); } lastTime = time; return time; } function getSyncedTime() { if (!Querysub.isInSyncedCall()) { return Date.now(); } let setTime = getSetTime(); if (setTime !== undefined) { return setTime; } if (isNode()) { let watcher = proxyWatcher.getTriggeredWatcherMaybeUndefined(); if (watcher) { throw new Error(`Trying to access time in a serverside non-temporary watcher. Clientside this is allowed, as infinite loops (render every frame) makes sense. Serverside this is not allowed. Did you try to run a call-type operatin in a watcher? If you manually created a watcher, you might want to set "temporary: true" if you will be immediately disposing it. You almost might want to fork your write logic with setImmediate to detach this from your watcher, so your write can access a single non-changing time. In ${watcher.debugName}`); } return Date.now(); } initTimeLoop(); return atomic(timeData().time) || Date.now(); } function timeDelayed(interval: number) { let call = getCurrentCallAllowUndefined(); if (call) { return call.runAtTime.time; } initInterval(interval); atomic(timeData().intervals[interval]); return Date.now(); } let nestedSyncDedupeHack = new Map void>(); const onNestedSynced = measureWrap(function onNestedSynced(callback: () => void) { const watcher = proxyWatcher.getTriggeredWatcherMaybeUndefined(); if (!watcher) { callback(); return; } let key = watcher.debugName + "_" + callback.toString(); if (nestedSyncDedupeHack.has(key)) { nestedSyncDedupeHack.set(key, callback); return; } nestedSyncDedupeHack.set(key, callback); const getPendingWatcher = (): SyncWatcher | undefined => { let checkWatchers: SyncWatcher[] = []; let checkedWatchers = new Set(); checkWatchers.push(watcher); while (true) { let watcher = checkWatchers.shift(); if (!watcher) break; if (watcher.hasAnyUnsyncedAccesses()) return watcher; // Wait until it's had time to spawn children watchers if (watcher.syncRunCount === 0) return watcher; let writes = watcher.pendingWrites; for (let writePath of writes.keys()) { // Find any watchers of these writes let callbacks = clientWatcher.getWatchersForPath(writePath); for (let callback of callbacks) { let nestedWatcher = proxyWatcher.getWatcherForTrigger(callback.callback); if (!nestedWatcher) continue; if (checkedWatchers.has(nestedWatcher)) continue; checkedWatchers.add(nestedWatcher); checkWatchers.push(nestedWatcher); } } } return undefined; }; const checkAgain = () => { const pendingWatcher = getPendingWatcher(); if (!pendingWatcher) { callback(); nestedSyncDedupeHack.delete(key); } else { //console.log(`Waiting for ${pendingWatcher.debugName} to finish`); getPendingWatcher(); let waitPromise = clientWatcher.waitForTriggerFinished(); if (!waitPromise) { pendingWatcher.onAfterTriggered.push(checkAgain); } else { void waitPromise.finally(() => { checkAgain(); }); } } }; Querysub.onCommitFinished(() => { checkAgain(); }); }); // Returns a value between 0 and 1 function hashRandom(base: string, offset: number): number { let hashBuf = Buffer.from(sha256(base + "_" + offset), "hex"); let value0 = hashBuf.readUInt32BE(0); let value1 = hashBuf.readUInt16BE(4); return value0 / (2 ** 32) + value1 / (2 ** 48); } // NOTE: Use `SocketFunction.logMessages = true` to diagnose what specifically is using the bandwidth. setImmediate(() => { let uploadLatest = 0; let downloadLatest = 0; let uploadTotal = 0; let downloadTotal = 0; registerDynamicResource("network|TOTAL_UPLOAD", () => uploadTotal); registerDynamicResource("network|TOTAL_DOWNLOAD", () => downloadTotal); registerDynamicResource("network|LATEST_UPLOAD", () => { let value = uploadLatest; uploadLatest = 0; return value; }); registerDynamicResource("network|LATEST_DOWNLOAD", () => { let value = downloadLatest; downloadLatest = 0; return value; }); SocketFunction.trackMessageSizes.upload.push(size => { uploadLatest += size; uploadTotal += size; }); SocketFunction.trackMessageSizes.download.push(size => { downloadLatest += size; downloadTotal += size; }); }); setImmediate(async () => { // Import, so it registers addStatPeriodic await import("../5-diagnostics/nodeMetadata"); await import("../diagnostics/pathAuditer"); }); setImmediate(async () => { if (isNode()) { const { doRegisterNodeForMachineCleanup } = await import("../deployManager/machineSchema"); doRegisterNodeForMachineCleanup(); } }); registerGetCompressNetwork(() => Querysub.COMPRESS_NETWORK); (globalThis as any).Querysub = Querysub; import "../diagnostics/watchdog"; import "../diagnostics/trackResources"; import { formatNumber, formatTime } from "socket-function/src/formatting/format"; import { getCountPerPaint } from "../functional/onNextPaint"; import { addEpsilons } from "socket-function/src/bits"; import { blue, green, magenta } from "socket-function/src/formatting/logColors"; import { MachineController } from "../deployManager/machineController"; import { getRecords, setRecord } from "../-b-authorities/dnsAuthority"; import { testTCPIsListening } from "socket-function/src/networking"; import { getNodeId, getNodeIdLocation } from "socket-function/src/nodeCache"; import { onAllPredictionsFinished } from "../-0-hooks/hooks"; import { LOCAL_DOMAIN } from "../0-path-value-core/PathRouter"; import { authorityLookup } from "../0-path-value-core/AuthorityLookup"; import { encodeParentFilter } from "../0-path-value-core/hackedPackedPathParentFiltering"; import { AliveChecker, registerAliveChecker } from "../2-proxy/garbageCollection"; import { QuerysubController, anyPredictionsPending, flushDelayedFunctions, onCallPredict, waitUntilAllPredictionsFinish } from "./QuerysubController";