diff --git a/pods/fulltext/src/server.ts b/pods/fulltext/src/server.ts index 200b981e8e..7a30b5a508 100644 --- a/pods/fulltext/src/server.ts +++ b/pods/fulltext/src/server.ts @@ -133,13 +133,23 @@ export async function startIndexer ( const token = request.token ?? req.headers.authorization?.split(' ')[1] const decoded = decodeToken(token) // Just to be safe - ctx.info('search', { classes: request._classes, query: request.query, workspace: decoded.workspace }) - await ctx.with('search', {}, async (ctx) => { - const docs = await ctx.with('search', { workspace: decoded.workspace }, (ctx) => - manager.fulltextAdapter.search(ctx, decoded.workspace, request._classes, request.query, request.fullTextLimit) - ) - req.body = docs - }) + await ctx.with( + 'search', + {}, + async (ctx) => { + req.body = await manager.fulltextAdapter.search( + ctx, + decoded.workspace, + request._classes, + request.query, + request.fullTextLimit + ) + }, + { + workspace: decoded.workspace, + classes: request._classes + } + ) } catch (err: any) { Analytics.handleError(err) console.error(err) @@ -153,10 +163,11 @@ export async function startIndexer ( const request = req.request.body as FulltextSearch const token = request.token ?? req.headers.authorization?.split(' ')[1] const decoded = decodeToken(token) // Just to be safe - ctx.info('fulltext-search', { ...request.query, workspace: decoded.workspace }) - await ctx.with('full-text-search', {}, async (ctx) => { - const result = await ctx.with('searchFulltext', {}, (ctx) => - searchFulltext( + await ctx.with( + 'full-text-search', + {}, + async (ctx) => { + const result = await searchFulltext( ctx, decoded.workspace, manager.sysHierarchy, @@ -164,9 +175,12 @@ export async function startIndexer ( request.query, request.options ) - ) - req.body = result - }) + req.body = result + }, + { + workspace: decoded.workspace + } + ) } catch (err: any) { Analytics.handleError(err) console.error(err) diff --git a/pods/server/src/rpc.ts b/pods/server/src/rpc.ts index c694000f8f..56252d19ff 100644 --- a/pods/server/src/rpc.ts +++ b/pods/server/src/rpc.ts @@ -139,6 +139,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur async function withSession ( req: Request, res: ExpressResponse, + method: string, operation: ( ctx: ClientSessionCtx, session: Session, @@ -183,9 +184,15 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur } const rpc = transactorRpc - const rateLimit = await sessions.handleRPC(rpc.context, rpc.session, rpc.client, async (ctx, rateLimit) => { - await operation(ctx, rpc.session, rateLimit, token) - }) + const rateLimit = await sessions.handleRPC( + rpc.context, + rpc.session, + method, + rpc.client, + async (ctx, rateLimit) => { + await operation(ctx, rpc.session, rateLimit, token) + } + ) if (rateLimit !== undefined) { const { remaining, limit, reset, retryAfter } = rateLimit const retryHeaders: OutgoingHttpHeaders = { @@ -212,7 +219,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur } app.get('/api/v1/ping/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session, rateLimit) => { + void withSession(req, res, 'ping', async (ctx, session, rateLimit) => { await session.ping(ctx) await sendJson( req, @@ -228,7 +235,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) app.get('/api/v1/find-all/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session, rateLimit) => { + void withSession(req, res, 'findAll', async (ctx, session, rateLimit) => { const _class = req.query.class as Ref> const query = req.query.query !== undefined ? JSON.parse(req.query.query as string) : {} const options = req.query.options !== undefined ? JSON.parse(req.query.options as string) : {} @@ -242,7 +249,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) app.post('/api/v1/find-all/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session, rateLimit) => { + void withSession(req, res, 'findAll', async (ctx, session, rateLimit) => { const { _class, query, options }: any = (await retrieveJson(req)) ?? {} const result = await session.findAllRaw(ctx, _class, query, options) @@ -251,7 +258,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) app.post('/api/v1/tx/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session, rateLimit) => { + void withSession(req, res, 'tx', async (ctx, session, rateLimit) => { const tx: any = (await retrieveJson(req)) ?? {} if (tx._class === core.class.TxDomainEvent) { @@ -271,7 +278,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur * @deprecated Use /api/v1/tx/:workspaceIdd instead */ app.post('/api/v1/event/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session) => { + void withSession(req, res, 'domainRequest', async (ctx, session) => { const event: any = (await retrieveJson(req)) ?? {} const { result } = await session.domainRequestRaw(ctx, COMMUNICATION_DOMAIN, { @@ -282,14 +289,14 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) app.get('/api/v1/account/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session, rateLimit) => { + void withSession(req, res, 'account', async (ctx, session, rateLimit) => { const result = session.getRawAccount() await sendJson(req, res, result, rateLimitToHeaders(rateLimit)) }) }) app.get('/api/v1/load-model/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session, rateLimit) => { + void withSession(req, res, 'loadModel', async (ctx, session, rateLimit) => { const lastModelTx = parseInt((req.query.lastModelTx as string) ?? '0') const lastHash = req.query.lastHash as string const result = await session.loadModelRaw(ctx, lastModelTx, lastHash) @@ -322,7 +329,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) app.get('/api/v1/search-fulltext/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session, rateLimit) => { + void withSession(req, res, 'searchFulltext', async (ctx, session, rateLimit) => { const query: SearchQuery = { query: req.query.query as string, classes: req.query.classes !== undefined ? JSON.parse(req.query.classes as string) : undefined, @@ -337,7 +344,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) app.get('/api/v1/request/:domain/:operation/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session) => { + void withSession(req, res, 'domainRequest', async (ctx, session) => { const domain = req.params.domain as OperationDomain const operation = req.params.operation @@ -351,7 +358,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) app.post('/api/v1/request/:domain/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session) => { + void withSession(req, res, 'domainRequest', async (ctx, session) => { const domain = req.params.domain as OperationDomain const params = retrieveJson(req) const { result } = await session.domainRequestRaw(ctx, domain, params) @@ -360,7 +367,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur }) app.post('/api/v1/ensure-person/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session, rateLimit, token) => { + void withSession(req, res, 'ensurePerson', async (ctx, session, rateLimit, token) => { const { socialType, socialValue, firstName, lastName } = (await retrieveJson(req)) ?? {} const accountClient = getAccountClient(token) @@ -438,7 +445,7 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur // To use in non-js (rust) clients that can't link to @hcengineering/core app.get('/api/v1/generate-id/:workspaceId', (req, res) => { - void withSession(req, res, async (ctx, session, rateLimit) => { + void withSession(req, res, 'generateId', async (ctx, session, rateLimit) => { const result = { id: generateId() } await sendJson(req, res, result, rateLimitToHeaders(rateLimit)) }) diff --git a/pods/server/src/server_http.ts b/pods/server/src/server_http.ts index c01a9a52e3..34016d42db 100644 --- a/pods/server/src/server_http.ts +++ b/pods/server/src/server_http.ts @@ -357,7 +357,7 @@ export function startHttpServer ( }) res.end(JSON.stringify({ success: true })) }, - { file: name, contentType, workspace: wsIds.uuid } + { contentType, workspace: wsIds.uuid } ) } catch (err: any) { Analytics.handleError(err) diff --git a/server/account-service/src/index.ts b/server/account-service/src/index.ts index e03878e93e..1803ad4095 100644 --- a/server/account-service/src/index.ts +++ b/server/account-service/src/index.ts @@ -429,7 +429,7 @@ export function serveAccount (measureCtx: MeasureContext, brandings: BrandingMap ctx.res.writeHead(200, KEEP_ALIVE_HEADERS) ctx.res.end(body) }, - { ...request } + { method: request.method } ) }) diff --git a/server/collaborator/src/extensions/storage.ts b/server/collaborator/src/extensions/storage.ts index 213800b6a9..f3d9213a0a 100644 --- a/server/collaborator/src/extensions/storage.ts +++ b/server/collaborator/src/extensions/storage.ts @@ -169,9 +169,17 @@ export class StorageExtension implements Extension { const { ctx, adapter } = this.configuration try { - return await ctx.with('load-document', {}, (ctx) => { - return adapter.loadDocument(ctx, documentName, context) - }) + return await ctx.with( + 'load-document', + {}, + (ctx) => { + return adapter.loadDocument(ctx, documentName, context) + }, + { + workspace: context.wsIds.uuid, + documentName + } + ) } catch (err: any) { Analytics.handleError(err) ctx.error('failed to load document', { documentName, error: err }) @@ -220,11 +228,18 @@ export class StorageExtension implements Extension { const now = Date.now() try { - const currMarkup = await ctx.with('save-document', {}, (ctx) => - adapter.saveDocument(ctx, documentName, document, context, { - prev: () => this.markups.get(documentName) ?? {}, - curr: () => this.configuration.transformer.fromYdoc(document) - }) + const currMarkup = await ctx.with( + 'save-document', + {}, + (ctx) => + adapter.saveDocument(ctx, documentName, document, context, { + prev: () => this.markups.get(documentName) ?? {}, + curr: () => this.configuration.transformer.fromYdoc(document) + }), + { + workspace: context.wsIds.uuid, + documentName + } ) this.markups.set(documentName, currMarkup ?? {}) diff --git a/server/collaborator/src/storage/platform.ts b/server/collaborator/src/storage/platform.ts index a4094f2fd5..d1c03e1bfd 100644 --- a/server/collaborator/src/storage/platform.ts +++ b/server/collaborator/src/storage/platform.ts @@ -50,16 +50,24 @@ export class PlatformStorageAdapter implements CollabStorageAdapter { try { ctx.info('load document content', { documentName }) - const ydoc = await ctx.with('loadCollabYdoc', {}, (ctx) => { - return withRetry( - ctx, - this.retryCount, - () => { - return loadCollabYdoc(ctx, this.storage, wsIds, documentId) - }, - this.retryInterval - ) - }) + const ydoc = await ctx.with( + 'loadCollabYdoc', + {}, + (ctx) => { + return withRetry( + ctx, + this.retryCount, + () => { + return loadCollabYdoc(ctx, this.storage, wsIds, documentId) + }, + this.retryInterval + ) + }, + { + workspace: context.wsIds.uuid, + documentName + } + ) if (ydoc !== undefined) { ctx.info('loaded from storage', { documentName }) @@ -76,11 +84,19 @@ export class PlatformStorageAdapter implements CollabStorageAdapter { try { ctx.info('load document initial content', { documentName, content }) - const markup = await ctx.with('loadCollabJson', {}, (ctx) => { - return withRetry(ctx, 5, () => { - return loadCollabJson(ctx, this.storage, wsIds, content) - }) - }) + const markup = await ctx.with( + 'loadCollabJson', + {}, + (ctx) => { + return withRetry(ctx, 5, () => { + return loadCollabJson(ctx, this.storage, wsIds, content) + }) + }, + { + workspace: context.wsIds.uuid, + documentName + } + ) if (markup !== undefined) { const ydoc = markupToYDoc(markup, documentId.objectAttr) @@ -117,16 +133,24 @@ export class PlatformStorageAdapter implements CollabStorageAdapter { try { ctx.info('save document ydoc content', { documentName }) - await ctx.with('saveCollabYdoc', {}, (ctx) => { - return withRetry( - ctx, - this.retryCount, - () => { - return saveCollabYdoc(ctx, this.storage, wsIds, documentId, document) - }, - this.retryInterval - ) - }) + await ctx.with( + 'saveCollabYdoc', + {}, + (ctx) => { + return withRetry( + ctx, + this.retryCount, + () => { + return saveCollabYdoc(ctx, this.storage, wsIds, documentId, document) + }, + this.retryInterval + ) + }, + { + workspace: context.wsIds.uuid, + documentName + } + ) } catch (err: any) { Analytics.handleError(err) ctx.error('failed to save document ydoc content', { documentName, error: err }) @@ -146,9 +170,17 @@ export class PlatformStorageAdapter implements CollabStorageAdapter { try { ctx.info('save document content to platform', { documentName }) - return await ctx.with('save-to-platform', {}, (ctx) => { - return this.saveDocumentToPlatform(ctx, client, context, documentName, getMarkup) - }) + return await ctx.with( + 'save-to-platform', + {}, + (ctx) => { + return this.saveDocumentToPlatform(ctx, client, context, documentName, getMarkup) + }, + { + workspace: context.wsIds.uuid, + documentName + } + ) } finally { await client.close() } @@ -202,45 +234,63 @@ export class PlatformStorageAdapter implements CollabStorageAdapter { return } - const blobId = await ctx.with('saveCollabJson', {}, (ctx) => { - return withRetry( - ctx, - this.retryCount, - () => { - return saveCollabJson(ctx, this.storage, wsIds, documentId, markup.curr[objectAttr]) - }, - this.retryInterval - ) - }) + const blobId = await ctx.with( + 'saveCollabJson', + {}, + (ctx) => { + return withRetry( + ctx, + this.retryCount, + () => { + return saveCollabJson(ctx, this.storage, wsIds, documentId, markup.curr[objectAttr]) + }, + this.retryInterval + ) + }, + { + workspace: context.wsIds.uuid, + documentName + } + ) await ctx.with('update', {}, () => client.diffUpdate(current, { [objectAttr]: blobId })) - await ctx.with('activity', {}, () => { - const space = hierarchy.isDerived(current._class, core.class.Space) ? (current._id as Ref) : current.space + await ctx.with( + 'activity', + {}, + () => { + const space = hierarchy.isDerived(current._class, core.class.Space) + ? (current._id as Ref) + : current.space - const data: AttachedData = { - objectId, - objectClass, - action: 'update', - attributeUpdates: { - attrKey: objectAttr, - attrClass: core.class.TypeMarkup, - prevValue: prevMarkup, - set: [currMarkup], - added: [], - removed: [], - isMixin: hierarchy.isMixin(objectClass) + const data: AttachedData = { + objectId, + objectClass, + action: 'update', + attributeUpdates: { + attrKey: objectAttr, + attrClass: core.class.TypeMarkup, + prevValue: prevMarkup, + set: [currMarkup], + added: [], + removed: [], + isMixin: hierarchy.isMixin(objectClass) + } } + return client.addCollection( + activity.class.DocUpdateMessage, + space, + current._id, + current._class, + 'docUpdateMessages', + data + ) + }, + { + workspace: context.wsIds.uuid, + documentName } - return client.addCollection( - activity.class.DocUpdateMessage, - space, - current._id, - current._class, - 'docUpdateMessages', - data - ) - }) + ) return markup.curr } diff --git a/server/core/src/triggers.ts b/server/core/src/triggers.ts index 311f55cd7a..bba7f9f2a1 100644 --- a/server/core/src/triggers.ts +++ b/server/core/src/triggers.ts @@ -151,7 +151,7 @@ export class Triggers { ctx.error('error during async processing', { err }) } }, - { count: matches.length } + { count: matches.length, workspace: ctrl.workspace.uuid } ) } } diff --git a/server/core/src/types.ts b/server/core/src/types.ts index fa7f1db36a..39cdf2de64 100644 --- a/server/core/src/types.ts +++ b/server/core/src/types.ts @@ -754,6 +754,7 @@ export interface SessionManager { handleRPC: ( requestCtx: MeasureContext, service: S, + method: string, ws: ConnectionSocket, operation: (ctx: ClientSessionCtx, rateLimit?: RateLimitInfo) => Promise ) => Promise diff --git a/server/datalake/src/client.ts b/server/datalake/src/client.ts index c207bee450..d6cb3291ef 100644 --- a/server/datalake/src/client.ts +++ b/server/datalake/src/client.ts @@ -240,12 +240,18 @@ export class DatalakeClient { } if (size === undefined || size < 64 * 1024 * 1024) { - return await ctx.with('direct-upload', {}, (ctx) => - this.uploadWithFormData(ctx, workspace, objectName, stream, { ...params, size }) + return await ctx.with( + 'direct-upload', + {}, + (ctx) => this.uploadWithFormData(ctx, workspace, objectName, stream, { ...params, size }), + { workspace, objectName } ) } else { - return await ctx.with('multipart-upload', {}, (ctx) => - this.uploadWithMultipart(ctx, workspace, objectName, stream, { ...params, size }) + return await ctx.with( + 'multipart-upload', + {}, + (ctx) => this.uploadWithMultipart(ctx, workspace, objectName, stream, { ...params, size }), + { workspace, objectName } ) } } diff --git a/server/datalake/src/index.ts b/server/datalake/src/index.ts index 321b20270e..1884e6b09a 100644 --- a/server/datalake/src/index.ts +++ b/server/datalake/src/index.ts @@ -197,8 +197,11 @@ export class DatalakeService implements StorageAdapter { size } - const { etag } = await ctx.with('put', {}, (ctx) => - this.retry(ctx, () => this.client.putObject(ctx, wsIds.uuid, objectName, stream, params)) + const { etag } = await ctx.with( + 'put', + {}, + (ctx) => this.retry(ctx, () => this.client.putObject(ctx, wsIds.uuid, objectName, stream, params)), + { workspace: wsIds.uuid, objectName } ) return { diff --git a/server/indexer/src/indexer/indexer.ts b/server/indexer/src/indexer/indexer.ts index 10e9511918..bf9f2939bb 100644 --- a/server/indexer/src/indexer/indexer.ts +++ b/server/indexer/src/indexer/indexer.ts @@ -180,23 +180,33 @@ class ElasticPushQueue { await this.pushQueue.add(async () => { try { try { - await this.ctx.with('push-elastic', {}, () => - this.fulltextAdapter.updateMany(this.ctx, this.workspace.uuid, docs) + await this.ctx.with( + 'push-elastic', + {}, + () => this.fulltextAdapter.updateMany(this.ctx, this.workspace.uuid, docs), + { workspace: this.workspace.uuid } ) await this.control?.heartbeat() } catch (err: any) { Analytics.handleError(err) // Try to push one by one - await this.ctx.with('push-elastic-by-one', {}, async () => { - for (const d of docs) { - try { - await this.fulltextAdapter.update(this.ctx, this.workspace.uuid, d.id, d) - } catch (err2: any) { - Analytics.handleError(err2) + await this.ctx.with( + 'push-elastic-by-one', + {}, + async () => { + for (const d of docs) { + try { + await this.fulltextAdapter.update(this.ctx, this.workspace.uuid, d.id, d) + } catch (err2: any) { + Analytics.handleError(err2) + } } + }, + { + workspace: this.workspace.uuid } - }) + ) } } catch (err: any) { Analytics.handleError(err) @@ -275,68 +285,81 @@ export class FullTextIndexPipeline implements FullTextPipeline { let processed = 0 let processedCommunication = 0 let hasCards = false - await ctx.with('reindex domain', { domain }, async (ctx) => { - // Iterate over all domain documents and add appropriate entries - const allDocs = this.storage.rawFind(ctx, domain) - try { - let lastPrint = platformNow() - const pushQueue = new ElasticPushQueue(this.fulltextAdapter, this.workspace, ctx, control) - while (true) { - await control?.heartbeat() - const docs = await allDocs.find(ctx) - if (docs.length === 0) { - break - } - const byClass = groupByArray>>(docs, (it) => it._class) - - for (const [v, values] of byClass.entries()) { - if (!isClassIndexable(this.hierarchy, v, this.contexts)) { - // Skip non indexable classes - continue - } - if (!hasCards && this.hierarchy.isDerived(v, card.class.Card)) { - hasCards = true - } - - await this.indexDocuments(ctx, v, values, pushQueue) + await ctx.with( + 'reindex domain', + { domain }, + async (ctx) => { + // Iterate over all domain documents and add appropriate entries + const allDocs = this.storage.rawFind(ctx, domain) + try { + let lastPrint = platformNow() + const pushQueue = new ElasticPushQueue(this.fulltextAdapter, this.workspace, ctx, control) + while (true) { await control?.heartbeat() + const docs = await allDocs.find(ctx) + if (docs.length === 0) { + break + } + const byClass = groupByArray>>(docs, (it) => it._class) + + for (const [v, values] of byClass.entries()) { + if (!isClassIndexable(this.hierarchy, v, this.contexts)) { + // Skip non indexable classes + continue + } + if (!hasCards && this.hierarchy.isDerived(v, card.class.Card)) { + hasCards = true + } + + await this.indexDocuments(ctx, v, values, pushQueue) + await control?.heartbeat() + } + + processed += docs.length + + // Define the thresholds for logging + + // Find the next threshold to print + + const now = platformNow() + if (now - lastPrint > printThresholdMs) { + ctx.info('processed', { + processed, + elapsed: Math.round(now - lastPrint), + domain, + workspace: this.workspace.uuid + }) + lastPrint = now + } } - - processed += docs.length - - // Define the thresholds for logging - - // Find the next threshold to print - - const now = platformNow() - if (now - lastPrint > printThresholdMs) { - ctx.info('processed', { - processed, - elapsed: Math.round(now - lastPrint), - domain, - workspace: this.workspace.uuid - }) - lastPrint = now - } + await pushQueue.waitProcessing() + } catch (err: any) { + ctx.error('failed to restore index state', { err }) + } finally { + await allDocs.close() } - await pushQueue.waitProcessing() - } catch (err: any) { - ctx.error('failed to restore index state', { err }) - } finally { - await allDocs.close() + if (hasCards) { + await ctx.with( + 'reindex-communication', + {}, + async (ctx) => { + try { + const pushQueue = new ElasticPushQueue(this.fulltextAdapter, this.workspace, ctx, control) + processedCommunication = await this.indexCommunication(ctx, control, pushQueue) + await pushQueue.waitProcessing() + } catch (err: any) { + ctx.error('failed to restore index state', { err }) + } + }, + { workspace: this.workspace.uuid } + ) + } + }, + { + domain, + workspace: this.workspace.uuid } - if (hasCards) { - await ctx.with('reindex-communication', {}, async (ctx) => { - try { - const pushQueue = new ElasticPushQueue(this.fulltextAdapter, this.workspace, ctx, control) - processedCommunication = await this.indexCommunication(ctx, control, pushQueue) - await pushQueue.waitProcessing() - } catch (err: any) { - ctx.error('failed to restore index state', { err }) - } - }) - } - }) + ) ctx.info('reindex done', { domain, processed, processedCommunication }) } @@ -450,95 +473,100 @@ export class FullTextIndexPipeline implements FullTextPipeline { const indexedDoc = createIndexedDoc(doc, this.hierarchy.findAllMixins(doc), doc.space) await rateLimit.add(async () => { - await ctx.with('process-document', { _class: doc._class }, async (ctx) => { - try { - // Collect all indexable values - const attributes = getFullTextIndexableAttributes(this.hierarchy, doc._class) - const content = getContent(this.hierarchy, attributes, doc) + await ctx.with( + 'process-document', + { _class: doc._class }, + async (ctx) => { + try { + // Collect all indexable values + const attributes = getFullTextIndexableAttributes(this.hierarchy, doc._class) + const content = getContent(this.hierarchy, attributes, doc) - indexedDoc.fulltextSummary = '' + indexedDoc.fulltextSummary = '' - for (const [, v] of Object.entries(content)) { - if (v.attr.type._class === core.class.TypeBlob) { - await ctx.with('process-blob', {}, (ctx) => this.processBlob(ctx, v, doc, indexedDoc), { - attr: v.attr.name, - value: v.value - }) - continue - } - - if (v.attr.type._class === core.class.TypeCollaborativeDoc) { - await this.processCollaborativeDoc(ctx, v, indexedDoc) - continue - } - if ((isFullTextAttribute(v.attr) || v.attr.isCustom === true) && v.value !== undefined) { - if (v.attr.type._class === core.class.TypeMarkup) { - ctx.withSync('markup-to-json-text', {}, () => { - indexedDoc.fulltextSummary += '\n' + jsonToText(markupToJSON(v.value)) + for (const [, v] of Object.entries(content)) { + if (v.attr.type._class === core.class.TypeBlob) { + await ctx.with('process-blob', {}, (ctx) => this.processBlob(ctx, v, doc, indexedDoc), { + attr: v.attr.name, + value: v.value }) - } else { - indexedDoc.fulltextSummary += '\n' + v.value + continue } - continue - } - - if (isIndexedAttribute(v.attr)) { - // We need to put indexed attr in place - - // Check for content changes and collect update - const dKey = docKey(v.attr.name, v.attr.attributeOf) - if (dKey !== '_class') { - if (typeof v.value !== 'object') { - indexedDoc[dKey] = v.value + if (v.attr.type._class === core.class.TypeCollaborativeDoc) { + await this.processCollaborativeDoc(ctx, v, indexedDoc) + continue + } + if ((isFullTextAttribute(v.attr) || v.attr.isCustom === true) && v.value !== undefined) { + if (v.attr.type._class === core.class.TypeMarkup) { + ctx.withSync('markup-to-json-text', {}, () => { + indexedDoc.fulltextSummary += '\n' + jsonToText(markupToJSON(v.value)) + }) } else { - // We need to extract only values - indexedDoc[dKey] = extractValues(v.value) + indexedDoc.fulltextSummary += '\n' + v.value } + + continue + } + + if (isIndexedAttribute(v.attr)) { + // We need to put indexed attr in place + + // Check for content changes and collect update + const dKey = docKey(v.attr.name, v.attr.attributeOf) + if (dKey !== '_class') { + if (typeof v.value !== 'object') { + indexedDoc[dKey] = v.value + } else { + // We need to extract only values + indexedDoc[dKey] = extractValues(v.value) + } + } + continue } - continue } - } - // trim to large content - if (indexedDoc.fulltextSummary.length > textLimit) { - indexedDoc.fulltextSummary = indexedDoc.fulltextSummary.slice(0, textLimit) - } + // trim to large content + if (indexedDoc.fulltextSummary.length > textLimit) { + indexedDoc.fulltextSummary = indexedDoc.fulltextSummary.slice(0, textLimit) + } - if (searchPresenter !== undefined) { - await ctx.with('update-search-presenter', { _class: doc._class }, async () => { - if (parentDocs === undefined) { - parentDocs = this.hierarchy.isDerived(_class, core.class.AttachedDoc) - ? await this.findParents(ctx, docs as unknown as AttachedDoc[]) - : undefined - } - const parentDoc = parentDocs?.get((doc as AttachedDoc).attachedTo) - if (spaceDocs === undefined) { - await updateSpaces() - } - const spaceDoc = spaceDocs?.get(doc.space) // docState.$lookup?.space - await updateDocWithPresenter(this.hierarchy, doc, indexedDoc, parentDoc, spaceDoc, searchPresenter) + if (searchPresenter !== undefined) { + await ctx.with('update-search-presenter', { _class: doc._class }, async () => { + if (parentDocs === undefined) { + parentDocs = this.hierarchy.isDerived(_class, core.class.AttachedDoc) + ? await this.findParents(ctx, docs as unknown as AttachedDoc[]) + : undefined + } + const parentDoc = parentDocs?.get((doc as AttachedDoc).attachedTo) + if (spaceDocs === undefined) { + await updateSpaces() + } + const spaceDoc = spaceDocs?.get(doc.space) // docState.$lookup?.space + await updateDocWithPresenter(this.hierarchy, doc, indexedDoc, parentDoc, spaceDoc, searchPresenter) + }) + } + + indexedDoc.id = doc._id + indexedDoc.space = doc.space + + if (this.listener?.onIndexing !== undefined) { + await this.listener.onIndexing(indexedDoc) + } + await pushQueue.push(indexedDoc) + } catch (err: any) { + ctx.error('failed to process document', { + id: doc._id, + class: doc._class, + workspace: this.workspace.uuid, + err: err.message, + stack: err.stack }) + Analytics.handleError(err) } - - indexedDoc.id = doc._id - indexedDoc.space = doc.space - - if (this.listener?.onIndexing !== undefined) { - await this.listener.onIndexing(indexedDoc) - } - await pushQueue.push(indexedDoc) - } catch (err: any) { - ctx.error('failed to process document', { - id: doc._id, - class: doc._class, - workspace: this.workspace.uuid, - err: err.message, - stack: err.stack - }) - Analytics.handleError(err) - } - }) + }, + { workspace: this.workspace.uuid } + ) }) } await rateLimit.waitProcessing() diff --git a/server/middleware/src/contextName.ts b/server/middleware/src/contextName.ts index c91c49bed8..1fb3f0ecef 100644 --- a/server/middleware/src/contextName.ts +++ b/server/middleware/src/contextName.ts @@ -41,16 +41,14 @@ export class ContextNameMiddleware extends BaseMiddleware implements Middleware } domainRequest (ctx: MeasureContext, domain: OperationDomain, params: DomainParams): Promise { - return ctx.with('domain-request', { source: ctx.contextData.service, domain }, (ctx) => { - return ctx.with( - `${domain}-${Object.keys(params)[0]}`, - {}, - async (ctx) => await this.provideDomainRequest(ctx, domain, params), - { - params - } - ) - }) + return ctx.with( + `${domain}-${Object.keys(params)[0]}`, + {}, + (ctx) => this.provideDomainRequest(ctx, domain, params), + { + workspace: this.context.workspace.uuid + } + ) } async tx (ctx: MeasureContext, txes: Tx[]): Promise { diff --git a/server/server/src/sessionManager.ts b/server/server/src/sessionManager.ts index 86079d7d08..d268b0a0b4 100644 --- a/server/server/src/sessionManager.ts +++ b/server/server/src/sessionManager.ts @@ -952,7 +952,7 @@ export class TSessionManager implements SessionManager { // await communicationApi.closeSession(sessionRef.session.sessionId) if (user !== guestAccount && user !== systemAccountUuid) { await this.trySetStatus( - workspace.context, + workspace.context.newChild('status', {}), pipeline, sessionRef.session, false, @@ -1214,7 +1214,6 @@ export class TSessionManager implements SessionManager { ): Promise { // Calculate total number of clients const reqId = generateId() - const mode = 'request' const source = service.token.extra?.service ?? '🤦‍♂️user' const st = Date.now() @@ -1237,7 +1236,7 @@ export class TSessionManager implements SessionManager { return } if (request.id === -1 && request.method === 'hello') { - await requestCtx.with('handleHello', { mode, source }, (ctx) => + await requestCtx.with('🧨 handleHello', { source }, (ctx) => this.handleHello(request, service, ctx, workspace, ws, requestCtx) ) return @@ -1300,16 +1299,18 @@ export class TSessionManager implements SessionManager { await workspace.with(async (pipeline) => { await requestCtx.with( '🧨' + request.method, - { mode, source }, + { source, mode: 'websocket' }, (callTx) => f.apply(service, [ this.createOpContext(callTx, requestCtx, pipeline, request.id, service, ws, rateLimit), ...params ]), - { ...request, user: service.getUser, socialId: service.getRawAccount().primarySocialId }, { - meta: request.meta - } + user: service.getUser(), + socialId: service.getRawAccount().primarySocialId, + workspace: workspace.wsId.uuid + }, + { meta: request.meta } ) }) } catch (err: any) { @@ -1336,6 +1337,7 @@ export class TSessionManager implements SessionManager { async handleRPC( requestCtx: MeasureContext, service: S, + method: string, ws: ConnectionSocket, operation: (ctx: ClientSessionCtx, rateLimit: RateLimitInfo | undefined) => Promise ): Promise { @@ -1345,7 +1347,6 @@ export class TSessionManager implements SessionManager { return await Promise.resolve(rateLimitStatus) } - const mode = 'rpc' const source = service.token.extra?.service ?? '🤦‍♂️user' // Calculate total number of clients @@ -1366,7 +1367,7 @@ export class TSessionManager implements SessionManager { try { await workspace.with(async (pipeline) => { - await requestCtx.with('🧨 handleRequest', { mode, source }, (callTx) => + await requestCtx.with('🧨 ' + method, { source, mode: 'rpc' }, (callTx) => operation( this.createOpContext(callTx, requestCtx, pipeline, reqId, service, ws, rateLimitStatus), rateLimitStatus @@ -1470,8 +1471,8 @@ export class TSessionManager implements SessionManager { if (account.uuid !== guestAccount && account.uuid !== systemAccountUuid) { void workspace.with(async (pipeline) => { // We do not need to wait for set-status, just return session to client - await ctx - .with('set-status', {}, (ctx) => this.trySetStatus(ctx, pipeline, service, true, service.workspace.uuid)) + await workspace.context + .with('🧨 status', {}, (ctx) => this.trySetStatus(ctx, pipeline, service, true, service.workspace.uuid)) .catch(() => {}) }) }