mirror of
https://github.com/Chocobozzz/PeerTube.git
synced 2025-10-05 02:39:33 +02:00
Relax transaction level
We don't need serializable level
This commit is contained in:
parent
bbe4910247
commit
9373f571be
2 changed files with 25 additions and 33 deletions
|
@ -5,59 +5,59 @@ import { Model } from 'sequelize-typescript'
|
||||||
import { sequelizeTypescript } from '@server/initializers/database.js'
|
import { sequelizeTypescript } from '@server/initializers/database.js'
|
||||||
import { logger } from './logger.js'
|
import { logger } from './logger.js'
|
||||||
|
|
||||||
function retryTransactionWrapper <T, A, B, C, D> (
|
function retryTransactionWrapper<T, A, B, C, D> (
|
||||||
functionToRetry: (arg1: A, arg2: B, arg3: C, arg4: D) => Promise<T>,
|
functionToRetry: (arg1: A, arg2: B, arg3: C, arg4: D) => Promise<T>,
|
||||||
arg1: A,
|
arg1: A,
|
||||||
arg2: B,
|
arg2: B,
|
||||||
arg3: C,
|
arg3: C,
|
||||||
arg4: D,
|
arg4: D
|
||||||
): Promise<T>
|
): Promise<T>
|
||||||
|
|
||||||
function retryTransactionWrapper <T, A, B, C> (
|
function retryTransactionWrapper<T, A, B, C> (
|
||||||
functionToRetry: (arg1: A, arg2: B, arg3: C) => Promise<T>,
|
functionToRetry: (arg1: A, arg2: B, arg3: C) => Promise<T>,
|
||||||
arg1: A,
|
arg1: A,
|
||||||
arg2: B,
|
arg2: B,
|
||||||
arg3: C
|
arg3: C
|
||||||
): Promise<T>
|
): Promise<T>
|
||||||
|
|
||||||
function retryTransactionWrapper <T, A, B> (
|
function retryTransactionWrapper<T, A, B> (
|
||||||
functionToRetry: (arg1: A, arg2: B) => Promise<T>,
|
functionToRetry: (arg1: A, arg2: B) => Promise<T>,
|
||||||
arg1: A,
|
arg1: A,
|
||||||
arg2: B
|
arg2: B
|
||||||
): Promise<T>
|
): Promise<T>
|
||||||
|
|
||||||
function retryTransactionWrapper <T, A> (
|
function retryTransactionWrapper<T, A> (
|
||||||
functionToRetry: (arg1: A) => Promise<T>,
|
functionToRetry: (arg1: A) => Promise<T>,
|
||||||
arg1: A
|
arg1: A
|
||||||
): Promise<T>
|
): Promise<T>
|
||||||
|
|
||||||
function retryTransactionWrapper <T> (
|
function retryTransactionWrapper<T> (
|
||||||
functionToRetry: () => Promise<T> | Bluebird<T>
|
functionToRetry: () => Promise<T> | Bluebird<T>
|
||||||
): Promise<T>
|
): Promise<T>
|
||||||
|
|
||||||
function retryTransactionWrapper <T> (
|
function retryTransactionWrapper<T> (
|
||||||
functionToRetry: (...args: any[]) => Promise<T>,
|
functionToRetry: (...args: any[]) => Promise<T>,
|
||||||
...args: any[]
|
...args: any[]
|
||||||
): Promise<T> {
|
): Promise<T> {
|
||||||
return transactionRetryer<T>(callback => {
|
return transactionRetryer<T>(callback => {
|
||||||
functionToRetry.apply(null, args)
|
functionToRetry.apply(null, args)
|
||||||
.then((result: T) => callback(null, result))
|
.then((result: T) => callback(null, result))
|
||||||
.catch(err => callback(err))
|
.catch(err => callback(err))
|
||||||
})
|
|
||||||
.catch(err => {
|
|
||||||
logger.warn(`Cannot execute ${functionToRetry.name} with many retries.`, { err })
|
|
||||||
throw err
|
|
||||||
})
|
})
|
||||||
|
.catch(err => {
|
||||||
|
logger.warn(`Cannot execute ${functionToRetry.name || 'function'} with many retries.`, { err })
|
||||||
|
throw err
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
function transactionRetryer <T> (func: (err: any, data: T) => any) {
|
function transactionRetryer<T> (func: (err: any, data: T) => any) {
|
||||||
return new Promise<T>((res, rej) => {
|
return new Promise<T>((res, rej) => {
|
||||||
retry(
|
retry(
|
||||||
{
|
{
|
||||||
times: 5,
|
times: 5,
|
||||||
|
|
||||||
errorFilter: err => {
|
errorFilter: err => {
|
||||||
const willRetry = (err.name === 'SequelizeDatabaseError')
|
const willRetry = err.name === 'SequelizeDatabaseError'
|
||||||
logger.debug('Maybe retrying the transaction function.', { willRetry, err, tags: [ 'sql', 'retry' ] })
|
logger.debug('Maybe retrying the transaction function.', { willRetry, err, tags: [ 'sql', 'retry' ] })
|
||||||
return willRetry
|
return willRetry
|
||||||
}
|
}
|
||||||
|
@ -68,7 +68,7 @@ function transactionRetryer <T> (func: (err: any, data: T) => any) {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
function saveInTransactionWithRetries <T extends Pick<Model, 'save' | 'changed'>> (model: T) {
|
function saveInTransactionWithRetries<T extends Pick<Model, 'save' | 'changed'>> (model: T) {
|
||||||
const changedKeys = model.changed() || []
|
const changedKeys = model.changed() || []
|
||||||
|
|
||||||
return retryTransactionWrapper(() => {
|
return retryTransactionWrapper(() => {
|
||||||
|
@ -89,24 +89,24 @@ function saveInTransactionWithRetries <T extends Pick<Model, 'save' | 'changed'>
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
function resetSequelizeInstance <T> (instance: Model<T>) {
|
function resetSequelizeInstance<T> (instance: Model<T>) {
|
||||||
return instance.reload()
|
return instance.reload()
|
||||||
}
|
}
|
||||||
|
|
||||||
function filterNonExistingModels <T extends { hasSameUniqueKeysThan (other: T): boolean }> (
|
function filterNonExistingModels<T extends { hasSameUniqueKeysThan(other: T): boolean }> (
|
||||||
fromDatabase: T[],
|
fromDatabase: T[],
|
||||||
newModels: T[]
|
newModels: T[]
|
||||||
) {
|
) {
|
||||||
return fromDatabase.filter(f => !newModels.find(newModel => newModel.hasSameUniqueKeysThan(f)))
|
return fromDatabase.filter(f => !newModels.find(newModel => newModel.hasSameUniqueKeysThan(f)))
|
||||||
}
|
}
|
||||||
|
|
||||||
function deleteAllModels <T extends Pick<Model, 'destroy'>> (models: T[], transaction: Transaction) {
|
function deleteAllModels<T extends Pick<Model, 'destroy'>> (models: T[], transaction: Transaction) {
|
||||||
return Promise.all(models.map(f => f.destroy({ transaction })))
|
return Promise.all(models.map(f => f.destroy({ transaction })))
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
function runInReadCommittedTransaction <T> (fn: (t: Transaction) => Promise<T>) {
|
function runInReadCommittedTransaction<T> (fn: (t: Transaction) => Promise<T>) {
|
||||||
const options = { isolationLevel: Transaction.ISOLATION_LEVELS.READ_COMMITTED }
|
const options = { isolationLevel: Transaction.ISOLATION_LEVELS.READ_COMMITTED }
|
||||||
|
|
||||||
return sequelizeTypescript.transaction(options, t => fn(t))
|
return sequelizeTypescript.transaction(options, t => fn(t))
|
||||||
|
|
|
@ -1,8 +1,7 @@
|
||||||
import { RunnerJobState, RunnerJobStateType } from '@peertube/peertube-models'
|
import { RunnerJobState, RunnerJobStateType } from '@peertube/peertube-models'
|
||||||
import { retryTransactionWrapper } from '@server/helpers/database-utils.js'
|
import { runInReadCommittedTransaction } from '@server/helpers/database-utils.js'
|
||||||
import { logger, loggerTagsFactory } from '@server/helpers/logger.js'
|
import { logger, loggerTagsFactory } from '@server/helpers/logger.js'
|
||||||
import { RUNNER_JOBS } from '@server/initializers/constants.js'
|
import { RUNNER_JOBS } from '@server/initializers/constants.js'
|
||||||
import { sequelizeTypescript } from '@server/initializers/database.js'
|
|
||||||
import { MRunner, MRunnerJob } from '@server/types/models/runners/index.js'
|
import { MRunner, MRunnerJob } from '@server/types/models/runners/index.js'
|
||||||
import express from 'express'
|
import express from 'express'
|
||||||
|
|
||||||
|
@ -10,7 +9,7 @@ const lTags = loggerTagsFactory('runner')
|
||||||
|
|
||||||
const updatingRunner = new Set<number>()
|
const updatingRunner = new Set<number>()
|
||||||
|
|
||||||
function updateLastRunnerContact (req: express.Request, runner: MRunner) {
|
export function updateLastRunnerContact (req: express.Request, runner: MRunner) {
|
||||||
const now = new Date()
|
const now = new Date()
|
||||||
|
|
||||||
// Don't update last runner contact too often
|
// Don't update last runner contact too often
|
||||||
|
@ -24,15 +23,13 @@ function updateLastRunnerContact (req: express.Request, runner: MRunner) {
|
||||||
|
|
||||||
logger.debug('Updating last runner contact for %s', runner.name, lTags(runner.name))
|
logger.debug('Updating last runner contact for %s', runner.name, lTags(runner.name))
|
||||||
|
|
||||||
retryTransactionWrapper(() => {
|
return runInReadCommittedTransaction(async transaction => {
|
||||||
return sequelizeTypescript.transaction(async transaction => {
|
return runner.save({ transaction })
|
||||||
return runner.save({ transaction })
|
|
||||||
})
|
|
||||||
}).catch(err => logger.error('Cannot update last runner contact for %s', runner.name, { err, ...lTags(runner.name) }))
|
}).catch(err => logger.error('Cannot update last runner contact for %s', runner.name, { err, ...lTags(runner.name) }))
|
||||||
.finally(() => updatingRunner.delete(runner.id))
|
.finally(() => updatingRunner.delete(runner.id))
|
||||||
}
|
}
|
||||||
|
|
||||||
function runnerJobCanBeCancelled (runnerJob: MRunnerJob) {
|
export function runnerJobCanBeCancelled (runnerJob: MRunnerJob) {
|
||||||
const allowedStates = new Set<RunnerJobStateType>([
|
const allowedStates = new Set<RunnerJobStateType>([
|
||||||
RunnerJobState.PENDING,
|
RunnerJobState.PENDING,
|
||||||
RunnerJobState.PROCESSING,
|
RunnerJobState.PROCESSING,
|
||||||
|
@ -41,8 +38,3 @@ function runnerJobCanBeCancelled (runnerJob: MRunnerJob) {
|
||||||
|
|
||||||
return allowedStates.has(runnerJob.state)
|
return allowedStates.has(runnerJob.state)
|
||||||
}
|
}
|
||||||
|
|
||||||
export {
|
|
||||||
updateLastRunnerContact,
|
|
||||||
runnerJobCanBeCancelled
|
|
||||||
}
|
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue