WEBDIENSTE/huly-platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2026-09-13 21:27:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

qfix: Update traces send to opentelemetry (#9728)

Browse Source 
+ Minimize amount of data send to traces.
+ Join WebSocket and RPC calls into one flat hierarchy.

Signed-off-by: Andrey Sobolev <haiodo@gmail.com>
This commit is contained in:
Andrey Sobolev
2025-08-27 23:15:55 +07:00
committed by GitHub
parent 2659693c5a
commit ef0db5597e
13 changed files with 394 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pods/fulltext/src/server.ts
+28 -14
View File
@@ -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)
pods/server/src/rpc.ts
+22 -15
View File
@@ -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<Class<Doc>>
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))
})
pods/server/src/server_http.ts
+1 -1
View File
@@ -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)
server/account-service/src/index.ts
+1 -1
View File
@@ -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 }
)
})
server/collaborator/src/extensions/storage.ts
+23 -8
View File
@@ -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 ?? {})
server/collaborator/src/storage/platform.ts
+111 -61
View File
@@ -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<Space>) : current.space
await ctx.with(
'activity',
{},
() => {
const space = hierarchy.isDerived(current._class, core.class.Space)
? (current._id as Ref<Space>)
: current.space
const data: AttachedData<DocUpdateMessage> = {
objectId,
objectClass,
action: 'update',
attributeUpdates: {
attrKey: objectAttr,
attrClass: core.class.TypeMarkup,
prevValue: prevMarkup,
set: [currMarkup],
added: [],
removed: [],
isMixin: hierarchy.isMixin(objectClass)
const data: AttachedData<DocUpdateMessage> = {
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
}
server/core/src/triggers.ts
+1 -1
View File
@@ -151,7 +151,7 @@ export class Triggers {
ctx.error('error during async processing', { err })
}
},
{ count: matches.length }
{ count: matches.length, workspace: ctrl.workspace.uuid }
)
}
}
server/core/src/types.ts
+1
View File
@@ -754,6 +754,7 @@ export interface SessionManager {
handleRPC: <S extends Session>(
requestCtx: MeasureContext,
service: S,
method: string,
ws: ConnectionSocket,
operation: (ctx: ClientSessionCtx, rateLimit?: RateLimitInfo) => Promise<void>
) => Promise<RateLimitInfo | undefined>
server/datalake/src/client.ts
+10 -4
View File
@@ -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 }
)
}
}
server/datalake/src/index.ts
+5 -2
View File
@@ -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 {
server/indexer/src/indexer/indexer.ts
+171 -143
View File
@@ -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<Doc, Ref<Class<Doc>>>(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<Doc, Ref<Class<Doc>>>(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()
server/middleware/src/contextName.ts
+8 -10
View File
@@ -41,16 +41,14 @@ export class ContextNameMiddleware extends BaseMiddleware implements Middleware
}
domainRequest (ctx: MeasureContext, domain: OperationDomain, params: DomainParams): Promise<DomainResult> {
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<SessionData>, txes: Tx[]): Promise<TxMiddlewareResult> {
server/server/src/sessionManager.ts
+12 -11
View File
@@ -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<void> {
// 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<S>(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<S extends Session>(
requestCtx: MeasureContext,
service: S,
method: string,
ws: ConnectionSocket,
operation: (ctx: ClientSessionCtx, rateLimit: RateLimitInfo | undefined) => Promise<void>
): Promise<RateLimitInfo | undefined> {
@@ -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(() => {})
})
}