Refactor into broker

plugins/scaffolder-backend/package.json

@@ -53,7 +53,9 @@
     "helmet": "^4.0.0",
     "isomorphic-git": "^1.8.0",
     "jsonschema": "^1.2.6",
+    "luxon": "^1.25.0",
     "morgan": "^1.10.0",
+    "p-queue": "^6.3.0",
     "uuid": "^8.2.0",
     "winston": "^3.2.1",
     "yaml": "^1.10.0"

plugins/scaffolder-backend/src/scaffolder/tasks/Database.ts

@@ -13,23 +13,79 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { Task, Status } from './types';
+import { DbTaskRow, TaskSpec } from './types';
+import { v4 as uuid } from 'uuid';
+import { DateTime } from 'luxon';
 
-export class Database {
-  private readonly store = new Map<string, Task>();
+export interface Database {
+  get(taskId: string): Promise<DbTaskRow>;
+  // updateTask(task: Task): Promise<DbTaskRow>;
+  createTask(task: TaskSpec): Promise<DbTaskRow>;
+  claimTask(): Promise<DbTaskRow>;
+  heartBeat(runId: string): Promise<void>;
+}
 
-  get(taskId: string) {
-    return this.store.get(taskId);
+export class InMemoryDatabase implements Database {
+  private readonly store = new Map<string, DbTaskRow>();
+
+  async heartBeat(runId: string): Promise<void> {
+    let task: DbTaskRow | undefined;
+
+    for (const t of this.store.values()) {
+      if (t.runId === runId) {
+        task = t;
+      }
+    }
+
+    if (!task) {
+      throw new Error('No task with matching runId found');
+    }
+
+    this.store.set(task.taskId, {
+      ...task,
+      lastHeartbeat: DateTime.local().toString(),
+    });
   }
 
-  write(task: Task) {
-    return task;
+  async claimTask(): Promise<DbTaskRow> {
+    for (const t of this.store.values()) {
+      if (t.status === 'OPEN') {
+        const task: DbTaskRow = {
+          ...t,
+          status: 'PROCESSING',
+          runId: uuid(),
+        };
+        this.store.set(t.taskId, task);
+        return task;
+      }
+    }
+    throw new Error('No task found');
   }
 
-  writeStatus(taskId: string, status: Status) {
+  async createTask(spec: TaskSpec): Promise<DbTaskRow> {
+    return {
+      taskId: uuid(),
+      spec,
+      status: 'OPEN',
+      retryCount: 0,
+      createdAt: new Date().toISOString(),
+    };
+  }
+
+  async get(taskId: string): Promise<DbTaskRow> {
     const task = this.store.get(taskId);
     if (task) {
-      this.store.set(taskId, { ...task, status });
+      return task;
     }
+    throw new Error(`could not found task ${taskId}`);
   }
+
+  // async updateTask(task: Task): Promise<Task> {
+  //   if (!task.taskId) {
+  //     throw new Error('Task must contain id');
+  //   }
+
+  //   this.store.set(task.taskId, task);
+  //   return task;
+  // }
 }

plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2021 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { MemoryTaskBroker, TaskAgent } from './TaskBroker';
+
+describe('MemoryTaskBroker', () => {
+  it('should claim a dispatched work item', async () => {
+    const broker = new MemoryTaskBroker();
+
+    await broker.dispatch({});
+    await expect(broker.claim()).resolves.toEqual(expect.any(TaskAgent));
+  });
+
+  it('should wait for a dispatched work item', async () => {
+    const broker = new MemoryTaskBroker();
+
+    const promise = broker.claim();
+
+    await expect(Promise.race([promise, 'waiting'])).resolves.toBe('waiting');
+
+    await broker.dispatch({});
+    await expect(promise).resolves.toEqual(expect.any(TaskAgent));
+  });
+
+  it('should dispatch multiple items and claim them in order', async () => {
+    const broker = new MemoryTaskBroker();
+
+    await broker.dispatch({ name: 'a' });
+    await broker.dispatch({ name: 'b' });
+    await broker.dispatch({ name: 'c' });
+
+    const taskA = await broker.claim();
+    const taskB = await broker.claim();
+    const taskC = await broker.claim();
+    await expect(taskA).toEqual(expect.any(TaskAgent));
+    await expect(taskB).toEqual(expect.any(TaskAgent));
+    await expect(taskC).toEqual(expect.any(TaskAgent));
+    await expect(taskA.spec.name).toBe('a');
+    await expect(taskB.spec.name).toBe('b');
+    await expect(taskC.spec.name).toBe('c');
+  });
+
+  it('should complete a task', async () => {
+    const broker = new MemoryTaskBroker();
+
+    await broker.dispatch({});
+    const task = await broker.claim();
+    await task.complete('COMPLETED');
+  });
+});

plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2021 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import {
+  CompletedTaskState,
+  Task,
+  TaskSpec,
+  TaskBroker,
+  Status,
+} from './types';
+import { v4 as uuid } from 'uuid';
+import { InMemoryDatabase } from './database';
+
+export class TaskAgent implements Task {
+  private heartbeartInterval?: ReturnType<typeof setInterval>;
+
+  static create(state: TaskState, db: InMemoryDatabase) {
+    const agent = new TaskAgent(state, db);
+    agent.start();
+    return agent;
+  }
+
+  // Runs heartbeat internally
+  private constructor(
+    private readonly state: TaskState,
+    private readonly db: InMemoryDatabase,
+  ) {}
+
+  get spec() {
+    return this.state.spec;
+  }
+
+  async emitLog(message: string): Promise<void> {
+    throw new Error('Method not implemented.');
+  }
+
+  async complete(result: CompletedTaskState): Promise<void> {
+    this.state.status = result === 'FAILED' ? 'COMPLETED' : 'FAILED';
+  }
+
+  private start() {
+    this.heartbeartInterval = setInterval(() => {
+      const runId = 'iiiid';
+      this.db.heartBeat(runId);
+    }, 4269);
+  }
+}
+
+interface TaskState {
+  spec: TaskSpec;
+  status: Status;
+  runId: string | undefined;
+}
+
+function defer() {
+  let resolve = () => {};
+  const promise = new Promise<void>(_resolve => {
+    resolve = _resolve;
+  });
+  return { promise, resolve };
+}
+
+export class MemoryTaskBroker implements TaskBroker {
+  private readonly db = new InMemoryDatabase();
+  private readonly tasks = new Array<TaskState>();
+  private deferredDispatch = defer();
+
+  async claim(): Promise<Task> {
+    for (;;) {
+      const pendingTask = await this.db.claimTask();
+      if (pendingTask) {
+        return TaskAgent.create(pendingTask, this.db);
+      }
+
+      await this.waitForDispatch();
+    }
+  }
+
+  async dispatch(spec: TaskSpec): Promise<void> {
+    this.tasks.push({
+      spec,
+      status: 'OPEN',
+      runId: undefined,
+    });
+    this.signalDispatch();
+  }
+
+  private waitForDispatch() {
+    return this.deferredDispatch.promise;
+  }
+
+  private signalDispatch() {
+    this.deferredDispatch.resolve();
+    this.deferredDispatch = defer();
+  }
+}

plugins/scaffolder-backend/src/scaffolder/tasks/TaskObserver.ts

@@ -1,15 +0,0 @@
-/*
- * Copyright 2021 Spotify AB
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */

plugins/scaffolder-backend/src/scaffolder/tasks/taskWorker.ts

@@ -1,34 +0,0 @@
-/*
- * Copyright 2021 Spotify AB
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import { Status, ClaimResponse } from './types';
-import { Database } from './Database';
-
-export class TaskWorker {
-  static async fromConfig() {}
-
-  constructor(private readonly database: Database) {}
-
-  claim(): Promise<ClaimResponse> {
-    return Promise.resolve(undefined);
-  }
-
-  setStatus(taskId: string, status: Status) {
-    this.database.writeStatus(taskId, status);
-  }
-
-  heartbeat(runId: number) {}
-}

plugins/scaffolder-backend/src/scaffolder/tasks/tasksDispatcher.ts

@@ -1,29 +0,0 @@
-/*
- * Copyright 2021 Spotify AB
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import { Task } from './types';
-import { Database } from './Database';
-
-export class taskDispatcher {
-  static async fromConfig(config: { database: Database }) {}
-
-  constructor(private readonly db: Database) {}
-
-  dispatch(task: Task): Promise<string> {
-    this.db.write(task);
-    return Promise.resolve('uuid');
-  }
-}

plugins/scaffolder-backend/src/scaffolder/tasks/types.ts

@@ -14,17 +14,43 @@
  * limitations under the License.
  */
 
-export type Status = 'OPEN' | 'PROCESSING' | 'FAILED' | 'CANCELLED';
-export type Task = {
+export type Status =
+  | 'OPEN'
+  | 'PROCESSING'
+  | 'FAILED'
+  | 'CANCELLED'
+  | 'COMPLETED';
+
+export type CompletedTaskState = 'FAILED' | 'COMPLETED';
+
+export type DbTaskRow = {
   taskId: string;
-  metadata: string;
+  spec: TaskSpec;
   status: Status;
-  lastHeartbeat: string;
+  lastHeartbeat?: string;
   retryCount: number;
   createdAt: string;
+  runId?: string;
+};
+
+export type DbTaskEventRow = {
+  id: number;
+  runId?: number;
+  stageName: string;
+  createdAt: string;
 };
 
-export type ClaimResponse = {
-  runId: number;
-  task: Task;
+export type TaskSpec = {
+  metadata: string;
 };
+
+export interface Task {
+  spec: TaskSpec;
+  emitLog(message: string): Promise<void>;
+  complete(result: CompletedTaskState): Promise<void>;
+}
+
+export interface TaskBroker {
+  claim(): Promise<Task>;
+  dispatch(spec: TaskSpec): Promise<void>;
+}