wieerwill/platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2025-06-09 09:20:54 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

UBERF-8425: Improved pg/acc/ws error handling (#9144)

Browse Source 
Signed-off-by: Alexey Zinoviev <alexey.zinoviev@xored.com>
This commit is contained in:
Alexey Zinoviev 2025-06-02 10:54:48 +04:00 committed by GitHub
parent 635304c4af
commit 89f81fd14a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 200 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
models/contact/src/migration.ts
View File

@ -195,13 +195,13 @@ async function assignEmployeeRoles (client: MigrationClient): Promise<void> {
client.logger.log('assigning roles to employees...', {})
const wsMembers = await client.accountClient.getWorkspaceMembers()
const persons = await client.traverse<Person>(DOMAIN_CONTACT, {
const personsIterator = await client.traverse<Person>(DOMAIN_CONTACT, {
_class: contact.class.Person
})
try {
while (true) {
const docs = await persons.next(50)
const docs = await personsIterator.next(50)
if (docs === null || docs?.length === 0) {
break
}
@ -236,7 +236,7 @@ async function assignEmployeeRoles (client: MigrationClient): Promise<void> {
}
}
} finally {
await persons.close()
await personsIterator.close()
client.logger.log('finished assigning roles to employees...', {})
}
}
@ -402,7 +402,7 @@ async function ensureGlobalPersonsForLocalAccounts (client: MigrationClient): Pr
async function createUserProfiles (client: MigrationClient): Promise<void> {
client.logger.log('creating user profiles for persons...', {})
const persons = await client.traverse<Person>(DOMAIN_CONTACT, {
const personsIterator = await client.traverse<Person>(DOMAIN_CONTACT, {
_class: contact.class.Person,
profile: { $exists: false }
})
@ -418,7 +418,7 @@ async function createUserProfiles (client: MigrationClient): Promise<void> {
try {
while (true) {
const docs = await persons.next(200)
const docs = await personsIterator.next(200)
if (docs === null || docs?.length === 0) {
break
}
@ -452,7 +452,7 @@ async function createUserProfiles (client: MigrationClient): Promise<void> {
}
}
} finally {
await persons.close()
await personsIterator.close()
client.logger.log('finished creating user profiles for persons...', {})
}
}
@ -460,13 +460,14 @@ async function createUserProfiles (client: MigrationClient): Promise<void> {
async function fixSocialIdCase (client: MigrationClient): Promise<void> {
client.logger.log('Fixing social id case...', {})
const socialIds = await client.traverse<SocialIdentity>(DOMAIN_CHANNEL, {
const socialIdsIterator = await client.traverse<SocialIdentity>(DOMAIN_CHANNEL, {
_class: contact.class.SocialIdentity
})
let updated = 0
try {
while (true) {
const docs = await socialIds.next(200)
const docs = await socialIdsIterator.next(200)
if (docs === null || docs?.length === 0) {
break
}
@ -480,7 +481,10 @@ async function fixSocialIdCase (client: MigrationClient): Promise<void> {
}
}
}
} finally {
await socialIdsIterator.close()
client.logger.log('Finished fixing social id case. Total updated:', { updated })
}
}
export const contactOperation: MigrateOperation = {

4
models/core/src/migration.ts
View File

@ -270,6 +270,7 @@ export async function migrateBackupMixins (client: MigrationClient): Promise<voi
const txIterator = await client.traverse<TxMixin<Doc, AttachedDoc>>(DOMAIN_TX, { _class: core.class.TxMixin })
try {
while (true) {
const mixinOps = await txIterator.next(500)
if (mixinOps === null || mixinOps.length === 0) break
@ -296,6 +297,9 @@ export async function migrateBackupMixins (client: MigrationClient): Promise<voi
}
}
}
} finally {
await txIterator.close()
}
}
async function migrateCollaborativeDocsToJson (client: MigrationClient): Promise<void> {

82
server/account/src/collections/postgres/postgres.ts
View File

@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
import { Sql } from 'postgres'
import { Sql, TransactionSql } from 'postgres'
import {
type Data,
type Version,
@ -249,11 +249,12 @@ implements DbCollection<T> {
}
async unsafe (sql: string, values: any[], client?: Sql): Promise<any[]> {
if (this.options.withRetryClient !== undefined) {
if (client !== undefined) {
return await client.unsafe(sql, values)
} else if (this.options.withRetryClient !== undefined) {
return await this.options.withRetryClient((_client) => _client.unsafe(sql, values))
} else {
const _client = client ?? this.client
return await _client.unsafe(sql, values)
return await this.client.unsafe(sql, values)
}
}
@ -696,7 +697,7 @@ export class PostgresAccountDB implements AccountDB {
}
}
withRetry = async <T>(callback: (client: Sql) => Promise<T>): Promise<T> => {
withRetry = async <T>(callback: (client: TransactionSql) => Promise<T>): Promise<T> => {
let attempt = 0
let delay = this.retryOptions.initialDelayMs
@ -723,14 +724,16 @@ export class PostgresAccountDB implements AccountDB {
return (
err.code === '40001' || // Retry transaction
err.code === '55P03' || // Lock not available
err.code === 'CONNECTION_CLOSED' || // This error is thrown if the connection was closed without an error.
err.code === 'CONNECTION_DESTROYED' || // This error is thrown for any queries that were pending when the timeout to sql.end({ timeout: X }) was reached. If the DB client is being closed completely retry will result in CONNECTION_ENDED which is not retried so should be fine.
msg.includes('RETRY_SERIALIZABLE')
)
}
async createWorkspace (data: WorkspaceData, status: WorkspaceStatusData): Promise<WorkspaceUuid> {
return await this.client.begin(async (client) => {
const workspaceUuid = await this.workspace.insertOne(data, client)
await this.workspaceStatus.insertOne({ ...status, workspaceUuid }, client)
return await this.withRetry(async (rTx) => {
const workspaceUuid = await this.workspace.insertOne(data, rTx)
await this.workspaceStatus.insertOne({ ...status, workspaceUuid }, rTx)
return workspaceUuid
})
@ -742,8 +745,10 @@ export class PostgresAccountDB implements AccountDB {
}
async assignWorkspace (accountUuid: AccountUuid, workspaceUuid: WorkspaceUuid, role: AccountRole): Promise<void> {
await this
.client`INSERT INTO ${this.client(this.getWsMembersTableName())} (workspace_uuid, account_uuid, role) VALUES (${workspaceUuid}, ${accountUuid}, ${role})`
await this.withRetry(
async (rTx) =>
await rTx`INSERT INTO ${this.client(this.getWsMembersTableName())} (workspace_uuid, account_uuid, role) VALUES (${workspaceUuid}, ${accountUuid}, ${role})`
)
}
async batchAssignWorkspace (data: [AccountUuid, WorkspaceUuid, AccountRole][]): Promise<void> {
@ -756,41 +761,51 @@ export class PostgresAccountDB implements AccountDB {
VALUES ${placeholders}
`
await this.client.unsafe(sql, values)
await this.withRetry(async (rTx) => await rTx.unsafe(sql, values))
}
async unassignWorkspace (accountUuid: AccountUuid, workspaceUuid: WorkspaceUuid): Promise<void> {
await this
.client`DELETE FROM ${this.client(this.getWsMembersTableName())} WHERE workspace_uuid = ${workspaceUuid} AND account_uuid = ${accountUuid}`
await this.withRetry(
async (rTx) =>
await rTx`DELETE FROM ${this.client(this.getWsMembersTableName())} WHERE workspace_uuid = ${workspaceUuid} AND account_uuid = ${accountUuid}`
)
}
async updateWorkspaceRole (accountUuid: AccountUuid, workspaceUuid: WorkspaceUuid, role: AccountRole): Promise<void> {
await this
.client`UPDATE ${this.client(this.getWsMembersTableName())} SET role = ${role} WHERE workspace_uuid = ${workspaceUuid} AND account_uuid = ${accountUuid}`
await this.withRetry(
async (rTx) =>
await rTx`UPDATE ${this.client(this.getWsMembersTableName())} SET role = ${role} WHERE workspace_uuid = ${workspaceUuid} AND account_uuid = ${accountUuid}`
)
}
async getWorkspaceRole (accountUuid: AccountUuid, workspaceUuid: WorkspaceUuid): Promise<AccountRole | null> {
const res = await this
.client`SELECT role FROM ${this.client(this.getWsMembersTableName())} WHERE workspace_uuid = ${workspaceUuid} AND account_uuid = ${accountUuid}`
return await this.withRetry(async (rTx) => {
const res =
await rTx`SELECT role FROM ${this.client(this.getWsMembersTableName())} WHERE workspace_uuid = ${workspaceUuid} AND account_uuid = ${accountUuid}`
return res[0]?.role ?? null
})
}
async getWorkspaceRoles (accountUuid: AccountUuid): Promise<Map<WorkspaceUuid, AccountRole>> {
const res = await this
.client`SELECT workspace_uuid, role FROM ${this.client(this.getWsMembersTableName())} WHERE account_uuid = ${accountUuid}`
return await this.withRetry(async (rTx) => {
const res =
await rTx`SELECT workspace_uuid, role FROM ${this.client(this.getWsMembersTableName())} WHERE account_uuid = ${accountUuid}`
return new Map(res.map((it) => [it.workspace_uuid as WorkspaceUuid, it.role]))
})
}
async getWorkspaceMembers (workspaceUuid: WorkspaceUuid): Promise<WorkspaceMemberInfo[]> {
const res: any = await this
.client`SELECT account_uuid, role FROM ${this.client(this.getWsMembersTableName())} WHERE workspace_uuid = ${workspaceUuid}`
return await this.withRetry(async (rTx) => {
const res: any =
await rTx`SELECT account_uuid, role FROM ${this.client(this.getWsMembersTableName())} WHERE workspace_uuid = ${workspaceUuid}`
return res.map((p: any) => ({
person: p.account_uuid,
role: p.role
}))
})
}
async getAccountWorkspaces (accountUuid: AccountUuid): Promise<WorkspaceInfoWithStatus[]> {
@ -824,7 +839,8 @@ export class PostgresAccountDB implements AccountDB {
ORDER BY s.last_visit DESC
`
const res: any = await this.client.unsafe(sql, [accountUuid])
return await this.withRetry(async (rTx) => {
const res: any = await rTx.unsafe(sql, [accountUuid])
for (const row of res) {
row.created_on = convertTimestamp(row.created_on)
@ -833,6 +849,7 @@ export class PostgresAccountDB implements AccountDB {
}
return convertKeysToCamelCase(res)
})
}
async getPendingWorkspace (
@ -929,31 +946,34 @@ export class PostgresAccountDB implements AccountDB {
// Note: SKIP LOCKED is supported starting from Postgres 9.5 and CockroachDB v22.2.1
sqlChunks.push('FOR UPDATE SKIP LOCKED')
return await this.withRetry(async (rTx) => {
// We must have all the conditions in the DB query and we cannot filter anything in the code
// because of possible concurrency between account services.
let res: any | undefined
await this.client.begin(async (client) => {
res = await client.unsafe(sqlChunks.join(' '), values)
const res: any = await rTx.unsafe(sqlChunks.join(' '), values)
if ((res.length ?? 0) > 0) {
await client.unsafe(
await rTx.unsafe(
`UPDATE ${this.workspaceStatus.getTableName()} SET processing_attempts = processing_attempts + 1, "last_processing_time" = $1 WHERE workspace_uuid = $2`,
[Date.now(), res[0].uuid]
)
}
})
return convertKeysToCamelCase(res[0]) as WorkspaceInfoWithStatus
})
}
async setPassword (accountUuid: AccountUuid, hash: Buffer, salt: Buffer): Promise<void> {
await this
.client`UPSERT INTO ${this.client(this.account.getPasswordsTableName())} (account_uuid, hash, salt) VALUES (${accountUuid}, ${hash.buffer as any}::bytea, ${salt.buffer as any}::bytea)`
await this.withRetry(
async (rTx) =>
await rTx`UPSERT INTO ${this.client(this.account.getPasswordsTableName())} (account_uuid, hash, salt) VALUES (${accountUuid}, ${hash.buffer as any}::bytea, ${salt.buffer as any}::bytea)`
)
}
async resetPassword (accountUuid: AccountUuid): Promise<void> {
await this
.client`DELETE FROM ${this.client(this.account.getPasswordsTableName())} WHERE account_uuid = ${accountUuid}`
await this.withRetry(
async (rTx) =>
await rTx`DELETE FROM ${this.client(this.account.getPasswordsTableName())} WHERE account_uuid = ${accountUuid}`
)
}
protected getMigrations (): [string, string][] {

8
server/account/src/utils.ts
View File

@ -66,7 +66,11 @@ import { isAdminEmail } from './admin'
export const GUEST_ACCOUNT = 'b6996120-416f-49cd-841e-e4a5d2e49c9b'
export const READONLY_GUEST_ACCOUNT = '83bbed9a-0867-4851-be32-31d49d1d42ce'
export async function getAccountDB (uri: string, dbNs?: string): Promise<[AccountDB, () => void]> {
export async function getAccountDB (
uri: string,
dbNs?: string,
appName: string = 'account'
): Promise<[AccountDB, () => void]> {
const isMongo = uri.startsWith('mongodb://')
if (isMongo) {
@ -85,7 +89,7 @@ export async function getAccountDB (uri: string, dbNs?: string): Promise<[Accoun
} else {
setDBExtraOptions({
connection: {
application_name: 'account'
application_name: appName
}
})
const client = getDBClient(sharedPipelineContextVars, uri)

16
server/postgres/src/storage.ts
View File

@ -223,7 +223,7 @@ class ConnectionMgr {
await client.execute('ROLLBACK;')
console.error({ message: 'failed to process tx', error: err.message, cause: err })
if (err.code !== '40001' || tries === maxTries) {
if (!this.isRetryableError(err) || tries === maxTries) {
return err
} else {
console.log('Transaction failed. Retrying.')
@ -267,7 +267,7 @@ class ConnectionMgr {
return { result: await fn(client) }
} catch (err: any) {
console.error({ message: 'failed to process sql', error: err.message, cause: err })
if (err.code !== '40001' || tries === maxTries) {
if (!this.isRetryableError(err) || tries === maxTries) {
return err
} else {
console.log('Read Transaction failed. Retrying.')
@ -330,6 +330,18 @@ class ConnectionMgr {
}
return conn
}
private isRetryableError (err: any): boolean {
const msg: string = err?.message ?? ''
return (
err.code === '40001' || // Retry transaction
err.code === '55P03' || // Lock not available
err.code === 'CONNECTION_CLOSED' || // This error is thrown if the connection was closed without an error.
err.code === 'CONNECTION_DESTROYED' || // This error is thrown for any queries that were pending when the timeout to sql.end({ timeout: X }) was reached. If the DB client is being closed completely retry will result in CONNECTION_ENDED which is not retried so should be fine.
msg.includes('RETRY_SERIALIZABLE')
)
}
}
class ValuesVariables {

2
server/tool/src/index.ts
View File

@ -323,7 +323,7 @@ export async function upgradeModel (
modelDb,
pipeline,
async (value) => {
await progress(90 + (Math.min(value, 100) / 100) * 10)
await progress(10 + (Math.min(value, 100) / 100) * 10)
},
wsIds.uuid
)

22
server/workspace-service/src/service.ts
View File

@ -52,6 +52,7 @@ import {
createPostgreeDestroyAdapter,
createPostgresAdapter,
createPostgresTxAdapter,
setDBExtraOptions,
shutdownPostgres
} from '@hcengineering/postgres'
import { doBackupWorkspace, doRestoreWorkspace } from '@hcengineering/server-backup'
@ -107,6 +108,7 @@ export type WorkspaceOperation = 'create' | 'upgrade' | 'all' | 'all+backup'
export class WorkspaceWorker {
runningTasks: number = 0
resolveBusy: (() => void) | null = null
id = randomUUID().slice(-8)
constructor (
readonly workspaceQueue: PlatformQueueProducer<QueueWorkspaceMessage>,
@ -146,6 +148,7 @@ export class WorkspaceWorker {
this.wakeup = this.defaultWakeup
const token = generateToken(systemAccountUuid, undefined, { service: 'workspace' })
ctx.info(`Starting workspace service worker ${this.id} with limit ${this.limit}...`)
ctx.info('Sending a handshake to the account service...')
const accountClient = getAccountClient(this.accountsUrl, token)
@ -158,12 +161,14 @@ export class WorkspaceWorker {
)
break
} catch (err: any) {
ctx.error('error', { err })
ctx.error('error during handshake', { err })
}
}
ctx.info('Successfully connected to the account service')
setDBExtraOptions({ connection: { application_name: `workspace-${this.id}` } })
registerTxAdapterFactory('mongodb', createMongoTxAdapter)
registerAdapterFactory('mongodb', createMongoAdapter)
registerDestroyFactory('mongodb', createMongoDestroyAdapter)
@ -187,6 +192,7 @@ export class WorkspaceWorker {
}
})
if (workspace == null) {
// no workspaces available, sleep before another attempt
await this.doSleep(ctx, opt)
} else {
void this.exec(async () => {
@ -196,9 +202,11 @@ export class WorkspaceWorker {
await this.doWorkspaceOperation(opContext, workspace, opt)
} catch (err: any) {
Analytics.handleError(err)
ctx.error('error', { err })
opContext.error('error', { err })
}
})
// sleep for a little bit to avoid bombarding the account service, also add jitter to avoid simultaneous requests from multiple workspace services
await new Promise((resolve) => setTimeout(resolve, Math.random() * 400 + 200))
}
}
}
@ -286,7 +294,7 @@ export class WorkspaceWorker {
await this.workspaceQueue.send(ws.uuid, [workspaceEvents.created()])
} catch (err: any) {
await opt.errorHandler(ws, err)
void opt.errorHandler(ws, err)
logger.log('error', err)
@ -385,7 +393,7 @@ export class WorkspaceWorker {
})
await this.workspaceQueue.send(ws.uuid, [workspaceEvents.upgraded()])
} catch (err: any) {
await opt.errorHandler(ws, err)
void opt.errorHandler(ws, err)
logger.log('error', err)
@ -720,8 +728,10 @@ export class WorkspaceWorker {
resolve()
this.wakeup = this.defaultWakeup
}
// sleep for 5 seconds for the next operation, or until a wakeup event
const sleepHandle = setTimeout(wakeup, opt.waitTimeout)
// sleep for N (5 by default) seconds for the next operation, or until a wakeup event
// add jitter to avoid simultaneous requests from multiple workspace services
const maxJitter = opt.waitTimeout * 0.2
const sleepHandle = setTimeout(wakeup, opt.waitTimeout + Math.random() * maxJitter)
this.wakeup = () => {
clearTimeout(sleepHandle)

5
server/workspace-service/src/ws-operations.ts
View File

@ -168,7 +168,8 @@ export async function createWorkspace (
await handleWsEvent?.('create-done', version, 100, '')
} catch (err: any) {
await handleWsEvent?.('ping', version, 0, `Create failed: ${err.message}`)
void handleWsEvent?.('ping', version, 0, `Create failed: ${err.message}`)
throw err
} finally {
await pipeline.close()
await storageAdapter.close()
@ -350,7 +351,7 @@ export async function upgradeWorkspaceWith (
await handleWsEvent?.('upgrade-done', version, 100, '')
} catch (err: any) {
ctx.error('upgrade-failed', { message: err.message })
await handleWsEvent?.('ping', version, 0, `Upgrade failed: ${err.message}`)
void handleWsEvent?.('ping', version, 0, `Upgrade failed: ${err.message}`)
throw err
} finally {
clearInterval(updateProgressHandle)

8
services/github/model-github/src/migration.ts
View File

@ -160,7 +160,7 @@ async function migrateFixMissingDocSyncInfo (client: MigrationClient): Promise<v
[github.mixin.GithubProject]: { $exists: true }
})
for (const p of projects) {
const issues = await client.traverse<Issue>(
const issuesIterator = await client.traverse<Issue>(
DOMAIN_TASK,
{
_class: tracker.class.Issue,
@ -178,8 +178,9 @@ async function migrateFixMissingDocSyncInfo (client: MigrationClient): Promise<v
}
)
let counter = 0
try {
while (true) {
const docs = await issues.next(1000)
const docs = await issuesIterator.next(1000)
if (docs === null || docs.length === 0) {
break
}
@ -222,10 +223,13 @@ async function migrateFixMissingDocSyncInfo (client: MigrationClient): Promise<v
}
}
}
} finally {
await issuesIterator.close()
if (counter > 0) {
console.log('Created', counter, 'DocSyncInfos')
}
}
}
}
async function migrateRemoveGithubComponents (client: MigrationClient): Promise<void> {