chore: fully distributed linting

packages/distributed/src/queues/base-queue.ts

@@ -24,7 +24,6 @@ const DEFAULT_OPTIONS: QueueOptions<any, any, any> = {
   maxTasks: 2,
   initialProgress: { start: 0, end: 100, current: 0 },
   initialData: {},
-  progressThrottle: ms('5s'),
   maxProgressDelay: ms('30s')
 }
 
@@ -151,19 +150,19 @@ export class BaseTaskQueue<TInput, TData, TError> implements ITaskQueue<TInput,
   private _runTask = async (task: Task<TInput, TData, TError>) => {
     this._logger.debug(`task "${task.id}" is about to start.`)
 
-    const progressCb = async (progress: TaskProgress) => {
+    const progressCb = async (progress: TaskProgress, data?: TData) => {
       task.status = 'running'
       task.progress = progress
+      if (data) {
+        task.data = data
+      }
       await this._taskRepo.inTransaction(async (repo) => {
         return repo.set(task)
       }, 'progressCallback')
     }
-    const throttledCb = _.throttle(progressCb, this._options.progressThrottle)
 
     try {
-      const terminatedTask = await this._taskRunner.run(task, throttledCb)
-
-      throttledCb.flush()
+      const terminatedTask = await this._taskRunner.run(task, progressCb)
 
       await this._taskRepo.inTransaction(async (repo) => {
         return repo.set(terminatedTask)

packages/distributed/src/queues/typings.ts

@@ -1,10 +1,13 @@
 export type TaskTrx<TInput, TData, TError> = (repo: TaskRepository<TInput, TData, TError>) => Promise<void>
 
 export type TaskHandler<TInput, TData, TError> = (task: Task<TInput, TData, TError>) => Promise<void>
-export type ProgressCb = (progress: TaskProgress) => void
+export type ProgressCb<_TInput, TData, _TError> = (progress: TaskProgress, data?: TData) => void
 
 export type TaskRunner<TInput, TData, TError> = {
-  run: (task: Task<TInput, TData, TError>, progress: ProgressCb) => Promise<TerminatedTask<TInput, TData, TError>>
+  run: (
+    task: Task<TInput, TData, TError>,
+    progress: ProgressCb<TInput, TData, TError>
+  ) => Promise<TerminatedTask<TInput, TData, TError>>
   cancel: (task: Task<TInput, TData, TError>) => Promise<void>
 }
 
@@ -27,7 +30,7 @@ export type TaskState = {
 export type Task<TInput, TData, TError> = TaskState & {
   id: string
   input: TInput
-  data: Partial<TData>
+  data: TData
   error?: TaskError<TInput, TData, TError>
 }
 
@@ -63,9 +66,8 @@ export type TaskProgress = {
 export type QueueOptions<_TInput, TData, _TError> = {
   maxTasks: number
   initialProgress: TaskProgress
-  initialData: Partial<TData>
+  initialData: TData
   maxProgressDelay: number
-  progressThrottle: number
 }
 
 export type TaskQueue<TInput, _TData, _TError> = {

packages/nlu-server/src/api/index.ts

@@ -297,7 +297,7 @@ export const createAPI = async (options: APIOptions, app: Application, baseLogge
         seed: 0
       }
 
-      const modelId = await app.lintDataset(appId, trainInput)
+      const modelId = await app.startLinting(appId, trainInput)
 
       const resp: http.LintResponseBody = { success: true, modelId: NLUEngine.modelIdService.toString(modelId) }
       return res.send(resp)

packages/nlu-server/src/application/errors.ts

@@ -40,6 +40,14 @@ export class TrainingAlreadyStartedError extends ResponseError {
   }
 }
 
+export class LintingAlreadyStartedError extends ResponseError {
+  constructor(appId: string, modelId: ModelId) {
+    const stringId = modelIdService.toString(modelId)
+    const lintKey = `${appId}/${stringId}`
+    super(`Linting "${lintKey}" already started...`, 409)
+  }
+}
+
 export class LangServerCommError extends ResponseError {
   constructor(err: Error) {
     const { message } = err

packages/nlu-server/src/application/index.ts

@@ -5,17 +5,14 @@ import {
   TrainInput,
   ServerInfo,
   TrainingStatus,
-  LintingState,
-  DatasetIssue,
-  IssueCode,
-  LintingError
+  LintingState
 } from '@botpress/nlu-client'
 import { Engine, ModelId, modelIdService, errors as engineErrors } from '@botpress/nlu-engine'
 import Bluebird from 'bluebird'
 import _ from 'lodash'
+import { TrainingRepository, TrainingListener } from '../infrastructure'
 import { LintingRepository } from '../infrastructure/linting-repo'
 import { ModelRepository } from '../infrastructure/model-repo'
-import { TrainingRepository, TrainingListener } from '../infrastructure/training-repo/typings'
 import {
   ModelDoesNotExistError,
   TrainingNotFoundError,
@@ -25,6 +22,7 @@ import {
   DatasetValidationError,
   LintingNotFoundError
 } from './errors'
+import { LintingQueue } from './linting-queue'
 import { TrainingQueue } from './training-queue'
 
 export class Application {
@@ -35,6 +33,7 @@ export class Application {
     private _trainingRepo: TrainingRepository,
     private _lintingRepo: LintingRepository,
     private _trainingQueue: TrainingQueue,
+    private _lintingQueue: LintingQueue,
     private _engine: Engine,
     private _serverVersion: string,
     baseLogger: Logger
@@ -55,6 +54,7 @@ export class Application {
     await this._trainingRepo.initialize()
     await this._trainingQueue.initialize()
     await this._lintingRepo.initialize()
+    await this._lintingQueue.initialize()
   }
 
   public async teardown() {
@@ -243,57 +243,31 @@ You can increase your cache size by the CLI or config.
     return detectedLanguages
   }
 
-  private _lint = async (appId: string, modelId: ModelId, trainInput: TrainInput): Promise<void> => {
-    const stringId = modelIdService.toString(modelId)
-    const key = `${appId}/${stringId}`
-
-    let lintingState: LintingState = {
-      currentCount: 0,
-      totalCount: -1,
-      issues: [],
-      status: 'linting-pending'
-    }
-
-    try {
-      await this._engine.lint(key, trainInput, {
-        minSpeed: 'slow',
-        progressCallback: (currentCount: number, totalCount: number, issues: DatasetIssue<IssueCode>[]) => {
-          lintingState = {
-            currentCount,
-            totalCount,
-            issues,
-            status: 'linting'
-          }
-          return this._lintingRepo.set({ appId, modelId, ...lintingState })
-        }
-      })
-    } catch (thrown) {
-      const err = thrown instanceof Error ? thrown : new Error(`${thrown}`)
-      const error: LintingError = { message: err.message, stack: err.stack, type: 'internal' }
-      return this._lintingRepo.set({ appId, modelId, ...lintingState, error })
-    }
-
-    return this._lintingRepo.set({ appId, modelId, ...lintingState, status: 'done' })
-  }
-
-  public async lintDataset(appId: string, trainInput: TrainInput): Promise<ModelId> {
+  public async startLinting(appId: string, trainInput: TrainInput): Promise<ModelId> {
     const modelId = modelIdService.makeId({
       ...trainInput,
       specifications: this._engine.getSpecifications()
     })
 
     // unhandled promise to return asap
-    void this._lint(appId, modelId, trainInput)
+    await this._lintingQueue.queueLinting(appId, modelId, trainInput)
 
     return modelId
   }
 
   public async getLintingState(appId: string, modelId: ModelId): Promise<LintingState> {
-    const state = await this._lintingRepo.get({ appId, modelId })
-    if (!state) {
-      throw new LintingNotFoundError(appId, modelId)
+    const linting = await this._lintingRepo.get({ appId, modelId })
+    if (linting) {
+      const { status, error, currentCount, totalCount, issues } = linting
+      return { status, error, currentCount, totalCount, issues }
     }
-    return state
+
+    const { specificationHash: currentSpec } = this._getSpecFilter()
+    if (modelId.specificationHash !== currentSpec) {
+      throw new InvalidModelSpecError(modelId, currentSpec)
+    }
+
+    throw new LintingNotFoundError(appId, modelId)
   }
 
   private _getSpecFilter = (): { specificationHash: string } => {

packages/nlu-server/src/application/linting-queue/index.ts

@@ -0,0 +1,96 @@
+import { queues } from '@botpress/distributed'
+import { Logger } from '@botpress/logger'
+import { TrainInput } from '@botpress/nlu-client'
+import * as NLUEngine from '@botpress/nlu-engine'
+import _ from 'lodash'
+import ms from 'ms'
+
+import { LintingRepository } from '../../infrastructure'
+import { LintingAlreadyStartedError, LintingNotFoundError } from '../errors'
+import { LintHandler } from './lint-handler'
+import { mapLintIdtoTaskId } from './lint-task-mapper'
+import { LintTaskRepo } from './lint-task-repo'
+import { LintTaskData, LintTaskError } from './typings'
+
+export const MIN_LINTING_HEARTBEAT = ms('10s')
+export const PROGRESS_THROTTLE = MIN_LINTING_HEARTBEAT / 2
+export const MAX_LINTING_HEARTBEAT = MIN_LINTING_HEARTBEAT * 3
+
+const TASK_OPTIONS: Partial<queues.QueueOptions<TrainInput, LintTaskData, LintTaskError>> = {
+  maxProgressDelay: MAX_LINTING_HEARTBEAT,
+  initialProgress: { start: 0, end: -1, current: 0 },
+  initialData: { issues: [] }
+}
+
+export type LintQueueOptions = {
+  maxLinting?: number
+}
+
+export abstract class LintingQueue {
+  constructor(private taskQueue: queues.TaskQueue<TrainInput, LintTaskData, LintTaskError>, private logger: Logger) {}
+
+  public initialize = this.taskQueue.initialize.bind(this.taskQueue)
+  public teardown = this.taskQueue.teardown.bind(this.taskQueue)
+  public getLocalLintingCount = this.taskQueue.getLocalTaskCount.bind(this.taskQueue)
+
+  public queueLinting = async (appId: string, modelId: NLUEngine.ModelId, trainInput: TrainInput) => {
+    try {
+      const taskId = mapLintIdtoTaskId({ modelId, appId })
+      await this.taskQueue.queueTask(taskId, trainInput)
+      this.logger.info(`[${taskId}] Linting Queued.`)
+    } catch (thrown) {
+      if (thrown instanceof queues.TaskAlreadyStartedError) {
+        throw new LintingAlreadyStartedError(appId, modelId)
+      }
+      throw thrown
+    }
+  }
+
+  public async cancelLinting(appId: string, modelId: NLUEngine.ModelId): Promise<void> {
+    try {
+      const taskId = mapLintIdtoTaskId({ modelId, appId })
+      await this.taskQueue.cancelTask(taskId)
+    } catch (thrown) {
+      if (thrown instanceof queues.TaskNotFoundError) {
+        throw new LintingNotFoundError(appId, modelId)
+      }
+      throw thrown
+    }
+  }
+}
+
+export class PgLintingQueue extends LintingQueue {
+  constructor(
+    pgURL: string,
+    lintingRepo: LintingRepository,
+    engine: NLUEngine.Engine,
+    logger: Logger,
+    opt: LintQueueOptions = {}
+  ) {
+    const lintTaskRepo = new LintTaskRepo(lintingRepo)
+    const lintHandler = new LintHandler(engine, logger.sub('linting-queue'))
+    const taskQueue = new queues.PGDistributedTaskQueue(pgURL, lintTaskRepo, lintHandler, logger, {
+      ...TASK_OPTIONS,
+      maxTasks: opt.maxLinting
+    })
+    super(taskQueue, logger)
+  }
+}
+
+export class LocalLintingQueue extends LintingQueue {
+  constructor(
+    lintingRepo: LintingRepository,
+    engine: NLUEngine.Engine,
+    baseLogger: Logger,
+    opt: LintQueueOptions = {}
+  ) {
+    const lintTaskRepo = new LintTaskRepo(lintingRepo)
+    const logger = baseLogger.sub('linting-queue')
+    const lintHandler = new LintHandler(engine, logger)
+    const taskQueue = new queues.LocalTaskQueue(lintTaskRepo, lintHandler, logger, {
+      ...TASK_OPTIONS,
+      maxTasks: opt.maxLinting
+    })
+    super(taskQueue, logger)
+  }
+}

packages/nlu-server/src/application/linting-queue/lint-handler.ts

@@ -0,0 +1,40 @@
+import { queues } from '@botpress/distributed'
+import { Logger } from '@botpress/logger'
+import { DatasetIssue, IssueCode, LintingError, TrainInput } from '@botpress/nlu-client'
+import * as NLUEngine from '@botpress/nlu-engine'
+import _ from 'lodash'
+import { PROGRESS_THROTTLE } from '.'
+import { mapLintErrorToTaskError } from './lint-task-mapper'
+import { LintTaskData, LintTaskError } from './typings'
+
+export class LintHandler implements queues.TaskRunner<TrainInput, LintTaskData, LintTaskError> {
+  constructor(private _engine: NLUEngine.Engine, private _logger: Logger) {}
+
+  public run = async (
+    task: queues.Task<TrainInput, LintTaskData, LintTaskError>,
+    progressCb: queues.ProgressCb<TrainInput, LintTaskData, LintTaskError>
+  ): Promise<queues.TerminatedTask<TrainInput, LintTaskData, LintTaskError>> => {
+    const throttledProgress = _.throttle(progressCb, PROGRESS_THROTTLE)
+
+    try {
+      await this._engine.lint(task.id, task.input, {
+        minSpeed: 'slow',
+        progressCallback: (currentCount: number, totalCount: number, issues: DatasetIssue<IssueCode>[]) => {
+          return throttledProgress({ start: 0, end: totalCount, current: currentCount }, { issues })
+        }
+      })
+      throttledProgress.flush()
+      this._logger.info(`[${task.id}] Linting Done.`)
+      return { ...task, status: 'done' }
+    } catch (thrown) {
+      throttledProgress.flush()
+      const err = thrown instanceof Error ? thrown : new Error(`${thrown}`)
+      const error: LintingError = { message: err.message, stack: err.stack, type: 'internal' }
+      return { ...task, status: 'errored', error: mapLintErrorToTaskError(error) }
+    }
+  }
+
+  public async cancel(task: queues.Task<TrainInput, LintTaskData, LintTaskError>): Promise<void> {
+    // TODO: make sure linting is cancellable
+  }
+}