Scheduled workflows

dbos-config.schema.json

@@ -145,6 +145,9 @@
         }
       }
     },
+    "skip_missed_scheduled_workflows": {
+      "type": "boolean"
+    },
     "application": {},
     "env": {},
     "version": {

migrations/20240430090000_tables.js

@@ -0,0 +1,13 @@
+exports.up = function(knex) {
+  return knex.schema.withSchema('dbos')
+    .createTable('scheduler_state', function(table) {
+        table.text('wf_function').notNullable();
+        table.bigInteger('last_wf_sched_time').notNullable();
+        table.primary(['wf_function']);
+    })
+};
+
+exports.down = function(knex) {
+  return knex.schema.withSchema('dbos')
+      .dropTableIfExists('scheduler_state');
+};

schemas/system_db_schema.ts

@@ -34,3 +34,8 @@ export interface workflow_inputs {
   workflow_uuid: string;
   inputs: string;
 }
+
+export interface scheduler_state {
+  wf_function: string;
+  last_wf_sched_time: number; // Time that has certainly been kicked off; others may have but OAOO will cover that
+}

src/dbos-executor.ts

@@ -60,6 +60,7 @@ export interface DBOSConfig {
   readonly application?: object;
   readonly debugProxy?: string;
   readonly debugMode?: boolean;
+  readonly skipMissedScheduledWorkflows?: boolean;
   readonly http?: {
     readonly cors_middleware?: boolean;
     readonly credentials?: boolean;
@@ -260,7 +261,7 @@ export class DBOSExecutor {
     for (const ro of registeredClassOperations) {
       if (ro.workflowConfig) {
         const wf = ro.registeredFunction as Workflow<any, any>;
-        this.#registerWorkflow(wf, ro.workflowConfig);
+        this.#registerWorkflow(wf, {registrationObject: ro, ...ro.workflowConfig});
         this.logger.debug(`Registered workflow ${ro.name}`);
       } else if (ro.txnConfig) {
         const tx = ro.registeredFunction as Transaction<any, any>;
@@ -432,6 +433,7 @@ export class DBOSExecutor {
 
     const runWorkflow = async () => {
       let result: R;
+
       // Execute the workflow.
       try {
         result = await wf(wCtxt, ...args);

src/dbos-runtime/config.ts

@@ -42,6 +42,7 @@ export interface ConfigFile {
   application: object;
   env: Record<string, string>;
   runtimeConfig?: DBOSRuntimeConfig;
+  skip_missed_scheduled_workflows?:boolean;
 }
 
 /*
@@ -195,6 +196,7 @@ export function parseConfigFile(cliOptions?: DBOSCLIStartOptions, useProxy: bool
     telemetry: configFile.telemetry || undefined,
     system_database: configFile.database.sys_db_name ?? `${poolConfig.database}_dbos_sys`,
     application: configFile.application || undefined,
+    skipMissedScheduledWorkflows: configFile.skip_missed_scheduled_workflows ?? false,
     env: configFile.env || {},
     http: configFile.http
   };

src/dbos-runtime/runtime.ts

@@ -7,6 +7,7 @@ import path from 'node:path';
 import { Server } from 'http';
 import { pathToFileURL } from 'url';
 import { DBOSKafka } from '../kafka/kafka';
+import { DBOSScheduler } from '../scheduler/scheduler';
 
 interface ModuleExports {
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -24,6 +25,7 @@ export class DBOSRuntime {
   private dbosExec: DBOSExecutor | null = null;
   private servers: { appServer: Server; adminServer: Server } | undefined;
   private kafka: DBOSKafka | null = null;
+  private scheduler: DBOSScheduler | null = null;
 
   constructor(dbosConfig: DBOSConfig, private readonly runtimeConfig: DBOSRuntimeConfig) {
     // Initialize workflow executor.
@@ -44,6 +46,10 @@ export class DBOSRuntime {
       this.dbosExec.logRegisteredHTTPUrls();
       this.kafka = new DBOSKafka(this.dbosExec);
       await this.kafka.initKafka();
+      this.kafka.logRegisteredKafkaEndpoints();
+      this.scheduler = new DBOSScheduler(this.dbosExec);
+      this.scheduler.initScheduler();
+      this.scheduler.logRegisteredSchedulerEndpoints();
     } catch (error) {
       this.dbosExec?.logger.error(error);
       if (error instanceof DBOSFailLoadOperationsError) {
@@ -97,6 +103,7 @@ export class DBOSRuntime {
    * Shut down the HTTP server and destroy workflow executor.
    */
   async destroy() {
+    await this.scheduler?.destroyScheduler();
     await this.kafka?.destroyKafka();
     if (this.servers) {
       this.servers.appServer.close();

src/foundationdb/fdb_system_database.ts

@@ -388,7 +388,15 @@ export class FoundationDBSystemDatabase implements SystemDatabase {
   }
 
 
-  async sleep(workflowUUID: string, functionID: number, durationSec: number): Promise<void> {
+  async sleep(_workflowUUID: string, _functionID: number, durationSec: number): Promise<void> {
     await sleep(durationSec * 1000); // TODO: Implement
   }
+
+  /* SCHEDULER */
+  getLastScheduledTime(_wfn: string): Promise<number | null> {
+    return Promise.resolve(null);
+  }
+  setLastScheduledTime(_wfn: string, _invtime: number): Promise<number | null> {
+    return Promise.resolve(null);
+  }
 }

src/index.ts

@@ -93,4 +93,10 @@ export {
 export {
   Kafka,
   KafkaConsume,
-} from "./kafka/kafka"
+} from "./kafka/kafka";
+
+export {
+  SchedulerConcurrencyMode,
+  SchedulerConfig,
+  Scheduled,
+} from "./scheduler/scheduler";

src/kafka/kafka.ts

@@ -82,17 +82,17 @@ export class DBOSKafka{
           throw new DBOSError(`Error registering method ${defaults.name}.${ro.name}: Kafka configuration not found. Does class ${defaults.name} have an @Kafka decorator?`)
         }
         const kafka = new KafkaJS(defaults.kafkaConfig);
-        const consumerConfig = ro.consumerConfig ?? { groupId: `dbos-kafka-group-${ro.kafkaTopic}`}
+        const consumerConfig = ro.consumerConfig ?? { groupId: `dbos-kafka-group-${ro.kafkaTopic}`};
         const consumer = kafka.consumer(consumerConfig);
-        await consumer.connect()
-        await consumer.subscribe({topic: ro.kafkaTopic, fromBeginning: true})
+        await consumer.connect();
+        await consumer.subscribe({topic: ro.kafkaTopic, fromBeginning: true});
         await consumer.run({
           eachMessage: async ({ topic, partition, message }) => {
             // This combination uniquely identifies a message for a given Kafka cluster
             const workflowUUID = `kafka-unique-id-${topic}-${partition}-${message.offset}`
             const wfParams = { workflowUUID: workflowUUID };
             // All operations annotated with Kafka decorators must take in these three arguments
-            const args: KafkaArgs = [topic, partition, message]
+            const args: KafkaArgs = [topic, partition, message];
             // We can only guarantee exactly-once-per-message execution of transactions and workflows.
             if (ro.txnConfig) {
               // Execute the transaction
@@ -113,4 +113,16 @@ export class DBOSKafka{
       await consumer.disconnect();
     }
   }
+
+  logRegisteredKafkaEndpoints() {
+    const logger = this.dbosExec.logger;
+    logger.info("Kafka endpoints supported:");
+    this.dbosExec.registeredOperations.forEach((registeredOperation) => {
+      const ro = registeredOperation as KafkaRegistration<unknown, unknown[], unknown>;
+      if (ro.kafkaTopic) {
+        const defaults = ro.defaults as KafkaDefaults;
+        logger.info(`    ${ro.kafkaTopic} -> ${defaults.name}.${ro.name}`);
+      }
+    });
+  }
 }