5.0.35

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@push.rocks/smartdata",
-  "version": "5.0.26",
+  "version": "5.0.35",
   "private": false,
   "description": "do more with data",
   "main": "dist_ts/index.js",
@@ -22,27 +22,26 @@
   },
   "homepage": "https://gitlab.com/pushrocks/smartdata#README",
   "dependencies": {
-    "@push.rocks/lik": "^6.0.2",
+    "@push.rocks/lik": "^6.0.14",
     "@push.rocks/smartdelay": "^3.0.1",
     "@push.rocks/smartlog": "^3.0.2",
     "@push.rocks/smartmongo": "^2.0.10",
     "@push.rocks/smartpromise": "^4.0.2",
-    "@push.rocks/smartrx": "^3.0.6",
-    "@push.rocks/smartstring": "^4.0.7",
-    "@push.rocks/smarttime": "^4.0.1",
-    "@push.rocks/smartunique": "^3.0.3",
-    "@push.rocks/taskbuffer": "^3.1.2",
-    "@tsclass/tsclass": "^4.0.42",
-    "mongodb": "^5.7.0"
+    "@push.rocks/smartrx": "^3.0.7",
+    "@push.rocks/smartstring": "^4.0.15",
+    "@push.rocks/smarttime": "^4.0.6",
+    "@push.rocks/smartunique": "^3.0.8",
+    "@push.rocks/taskbuffer": "^3.1.7",
+    "@tsclass/tsclass": "^4.0.52",
+    "mongodb": "^6.5.0"
   },
   "devDependencies": {
     "@gitzone/tsbuild": "^2.1.66",
     "@gitzone/tsrun": "^1.2.44",
     "@gitzone/tstest": "^1.0.77",
-    "@push.rocks/qenv": "^6.0.2",
-    "@push.rocks/tapbundle": "^5.0.15",
-    "@types/node": "^20.4.10",
-    "@types/shortid": "0.0.29"
+    "@push.rocks/qenv": "^6.0.5",
+    "@push.rocks/tapbundle": "^5.0.22",
+    "@types/node": "^20.11.30"
   },
   "files": [
     "ts/**/*",

test/test.distributedcoordinator.ts

@@ -27,37 +27,35 @@ tap.test('should instantiate DistributedClass', async (tools) => {
 tap.test('DistributedClass should update the time', async (tools) => {
     const distributedCoordinator = new SmartdataDistributedCoordinator(testDb);
     await distributedCoordinator.start();
-    console.log('test: started');
     const initialTime = distributedCoordinator.ownInstance.data.lastUpdated;
     await distributedCoordinator.sendHeartbeat();
-    console.log('test: sent heartbeat');
     const updatedTime = distributedCoordinator.ownInstance.data.lastUpdated;
     expect(updatedTime).toBeGreaterThan(initialTime);
     await distributedCoordinator.stop();
 });
 
 tap.test('should instantiate SmartdataDistributedCoordinator', async (tools) => {
-    const coordinator = new SmartdataDistributedCoordinator(testDb);
-    await coordinator.start();
-    expect(coordinator).toBeInstanceOf(SmartdataDistributedCoordinator);
-    await coordinator.stop();
+    const distributedCoordinator = new SmartdataDistributedCoordinator(testDb);
+    await distributedCoordinator.start();
+    expect(distributedCoordinator).toBeInstanceOf(SmartdataDistributedCoordinator);
+    await distributedCoordinator.stop();
 });
 
 tap.test('SmartdataDistributedCoordinator should update leader status', async (tools) => {
-    const coordinator = new SmartdataDistributedCoordinator(testDb);
-    await coordinator.start();
-    await coordinator.checkAndMaybeLead();
-    expect(coordinator.ownInstance.data.elected).toBeOneOf([true, false]);
-    await coordinator.stop();
-    process.exit(0);
+    const distributedCoordinator = new SmartdataDistributedCoordinator(testDb);
+    await distributedCoordinator.start();
+    await distributedCoordinator.checkAndMaybeLead();
+    expect(distributedCoordinator.ownInstance.data.elected).toBeOneOf([true, false]);
+    await distributedCoordinator.stop();
 });
 
 tap.test('SmartdataDistributedCoordinator should handle distributed task requests', async (tools) => {
-    const coordinator = new SmartdataDistributedCoordinator(testDb);
-    await coordinator.start();
+    const distributedCoordinator = new SmartdataDistributedCoordinator(testDb);
+    await distributedCoordinator.start();
 
     const mockTaskRequest: taskbuffer.distributedCoordination.IDistributedTaskRequest = {
         submitterId: "mockSubmitter12345",    // Some unique mock submitter ID
+        requestResponseId: 'uni879873462hjhfkjhsdf', // Some unique ID for the request-response
         taskName: "SampleTask",
         taskVersion: "1.0.0",                  // Assuming it's a version string
         taskExecutionTime: Date.now(),
@@ -66,18 +64,19 @@ tap.test('SmartdataDistributedCoordinator should handle distributed task request
         status: 'requesting'
     };
 
-    const response = await coordinator.fireDistributedTaskRequest(mockTaskRequest);
-    expect(response).toBeTruthy();  // based on your expected structure for the response
-    await coordinator.stop();
+    const response = await distributedCoordinator.fireDistributedTaskRequest(mockTaskRequest);
+    console.log(response)  // based on your expected structure for the response
+    await distributedCoordinator.stop();
 });
 
 tap.test('SmartdataDistributedCoordinator should update distributed task requests', async (tools) => {
-    const coordinator = new SmartdataDistributedCoordinator(testDb);
+    const distributedCoordinator = new SmartdataDistributedCoordinator(testDb);
     
-    await coordinator.start();
+    await distributedCoordinator.start();
 
     const mockTaskRequest: taskbuffer.distributedCoordination.IDistributedTaskRequest = {
         submitterId: "mockSubmitter12345",    // Some unique mock submitter ID
+        requestResponseId: 'uni879873462hjhfkjhsdf', // Some unique ID for the request-response
         taskName: "SampleTask",
         taskVersion: "1.0.0",                  // Assuming it's a version string
         taskExecutionTime: Date.now(),
@@ -87,18 +86,27 @@ tap.test('SmartdataDistributedCoordinator should update distributed task request
     };
     
 
-    await coordinator.updateDistributedTaskRequest(mockTaskRequest);
+    await distributedCoordinator.updateDistributedTaskRequest(mockTaskRequest);
     // Here, we can potentially check if a DB entry got updated or some other side-effect of the update method.
+    await distributedCoordinator.stop();
 });
 
 tap.test('should elect only one leader amongst multiple instances', async (tools) => {
     const coordinators = Array.from({ length: totalInstances }).map(() => new SmartdataDistributedCoordinator(testDb));
-    await Promise.all(coordinators.map(coordinator => coordinator.init()));
-
-    await Promise.all(coordinators.map(coordinator => coordinator.checkAndMaybeLead()));
-
+    await Promise.all(coordinators.map(coordinator => coordinator.start()));
     const leaders = coordinators.filter(coordinator => coordinator.ownInstance.data.elected);
+    for (const leader of leaders) {
+        console.log(leader.ownInstance);
+    }
     expect(leaders.length).toEqual(1);
+
+    // stopping clears a coordinator from being elected.
+    await Promise.all(coordinators.map(coordinator => coordinator.stop()));
 });
 
+tap.test('should clean up', async () => {
+    await smartmongoInstance.stopAndDumpToDir(`.nogit/testdata/`);
+    setTimeout(() => process.exit(), 2000);
+})
+
 tap.start({ throwOnError: true });

test/test.easystore.ts

@@ -26,7 +26,7 @@ tap.test('should create a testinstance as database', async () => {
 tap.skip.test('should connect to atlas', async (tools) => {
   const databaseName = `test-smartdata-${smartunique.shortId()}`;
   testDb = new smartdata.SmartdataDb({
-    mongoDbUrl: testQenv.getEnvVarOnDemand('MONGO_URL'),
+    mongoDbUrl: await testQenv.getEnvVarOnDemand('MONGO_URL'),
     mongoDbName: databaseName,
   });
   await testDb.init();

test/test.watch.ts

@@ -28,7 +28,7 @@ tap.test('should create a testinstance as database', async () => {
 tap.skip.test('should connect to atlas', async (tools) => {
   const databaseName = `test-smartdata-${smartunique.shortId()}`;
   testDb = new smartdata.SmartdataDb({
-    mongoDbUrl: testQenv.getEnvVarOnDemand('MONGO_URL'),
+    mongoDbUrl: await testQenv.getEnvVarOnDemand('MONGO_URL'),
     mongoDbName: databaseName,
   });
   await testDb.init();

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@push.rocks/smartdata',
-  version: '5.0.26',
+  version: '5.0.35',
   description: 'do more with data'
 }

ts/smartdata.classes.collection.ts

@@ -56,7 +56,7 @@ export const setDefaultManagerForDoc = <T>(managerArg: IManager, dbDocArg: T): T
  * This is a decorator that will tell the decorated class what dbTable to use
  * @param dbArg
  */
-export function Manager<TManager extends IManager>(managerArg?: TManager | TDelayed<TManager>) {
+export function managed<TManager extends IManager>(managerArg?: TManager | TDelayed<TManager>) {
   return function classDecorator<T extends { new (...args: any[]): any }>(constructor: T) {
     return class extends constructor {
       public static get collection() {
@@ -109,6 +109,11 @@ export function Manager<TManager extends IManager>(managerArg?: TManager | TDela
   };
 }
 
+/**
+ * @dpecrecated use @managed instead
+ */
+export const Manager = managed;
+
 export class SmartdataCollection<T> {
   /**
    * the collection that is used
@@ -269,7 +274,7 @@ export class SmartdataCollection<T> {
    * if this.objectValidation is not set it passes.
    */
   private checkDoc(docArg: T): Promise<void> {
-    const done = plugins.smartq.defer<void>();
+    const done = plugins.smartpromise.defer<void>();
     let validationResult = true;
     if (this.objectValidation) {
       validationResult = this.objectValidation(docArg);

ts/smartdata.classes.distributedcoordinator.ts

@@ -1,9 +1,10 @@
 import * as plugins from './smartdata.plugins.js';
 import { SmartdataDb } from './smartdata.classes.db.js';
-import { Manager, setDefaultManagerForDoc } from './smartdata.classes.collection.js';
+import { managed, setDefaultManagerForDoc } from './smartdata.classes.collection.js';
 import { SmartDataDbDoc, svDb, unI } from './smartdata.classes.doc.js';
+import { SmartdataDbWatcher } from './smartdata.classes.watcher.js';
 
-@Manager()
+@managed()
 export class DistributedClass extends SmartDataDbDoc<DistributedClass, DistributedClass> {
   // INSTANCE
   @unI()
@@ -11,8 +12,9 @@ export class DistributedClass extends SmartDataDbDoc<DistributedClass, Distribut
 
   @svDb()
   public data: {
-    status: 'bidding' | 'settled' | 'initializing' | 'stopped';
+    status: 'initializing' | 'bidding' | 'settled' | 'stopped';
     biddingShortcode?: string;
+    biddingStartTime?: number;
     lastUpdated: number;
     elected: boolean;
     /**
@@ -38,6 +40,7 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
   public db: SmartdataDb;
   private asyncExecutionStack = new plugins.lik.AsyncExecutionStack();
   public ownInstance: DistributedClass;
+  public distributedWatcher: SmartdataDbWatcher<DistributedClass>;
 
   constructor(dbArg: SmartdataDb) {
     super();
@@ -52,11 +55,20 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
   }
 
   public async stop() {
+    await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      if (this.distributedWatcher) {
+        await this.distributedWatcher.close();
+      }
       if (this.ownInstance?.data.elected) {
         this.ownInstance.data.elected = false;
-    } else {
-      console.log(`can't stop a distributed instance that has not been started yet.`);
       }
+      if (this.ownInstance?.data.status === 'stopped') {
+        console.log(`stopping a distributed instance that has not been started yet.`);
+      }
+      this.ownInstance.data.status = 'stopped';
+      await this.ownInstance.save();
+      console.log(`stopped ${this.ownInstance.id}`);
+    });
   }
 
   public id = plugins.smartunique.uni('distributedInstance');
@@ -64,15 +76,32 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
   private startHeartbeat = async () => {
     while (this.ownInstance.data.status !== 'stopped') {
       await this.sendHeartbeat();
-      await plugins.smartdelay.delayFor(10000);
+      await plugins.smartdelay.delayForRandom(5000, 10000);
     }
   };
 
   public async sendHeartbeat() {
     await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      if (this.ownInstance.data.status === 'stopped') {
+        console.log(`aborted sending heartbeat because status is stopped`);
+        return;
+      }
       await this.ownInstance.updateFromDb();
       this.ownInstance.data.lastUpdated = Date.now();
       await this.ownInstance.save();
+      console.log(`sent heartbeat for ${this.ownInstance.id}`);
+      const allInstances = DistributedClass.getInstances({});
+    });
+    if (this.ownInstance.data.status === 'stopped') {
+      console.log(`aborted sending heartbeat because status is stopped`);
+      return;
+    }
+    const eligibleLeader = await this.getEligibleLeader();
+    // not awaiting here because we don't want to block the heartbeat
+    this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      if (!eligibleLeader && this.ownInstance.data.status === 'settled') {
+        this.checkAndMaybeLead();
+      }
     });
   }
   private async init() {
@@ -90,6 +119,8 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
         };
         await this.ownInstance.save();
       });
+    } else {
+      console.warn(`distributed instance already initialized`);
     }
 
     // lets enable the heartbeat
@@ -101,42 +132,73 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
     return this.ownInstance;
   }
 
+  public async getEligibleLeader() {
+    return this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      const allInstances = await DistributedClass.getInstances({});
+      let leaders = allInstances.filter((instanceArg) => instanceArg.data.elected === true);
+      const eligibleLeader = leaders.find(
+        (leader) =>
+          leader.data.lastUpdated >=
+          Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ seconds: 20 })
+      );
+      return eligibleLeader;
+    });
+  }
+
   // --> leader election
   public async checkAndMaybeLead() {
-    const allInstances = await DistributedClass.getInstances({});
-    let leader = allInstances.find((instanceArg) => instanceArg.data.elected === true);
-    if (
-      leader &&
-      leader.data.lastUpdated >=
-        Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ minutes: 1 })
-    ) {
+    await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      this.ownInstance.data.status = 'initializing';
+      this.ownInstance.save();
+    });
+    if (await this.getEligibleLeader()) {
       await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
         await this.ownInstance.updateFromDb();
         this.ownInstance.data.status = 'settled';
         await this.ownInstance.save();
+        console.log(`${this.ownInstance.id} settled as follower`);
       });
       return;
+    } else if (
+      (await DistributedClass.getInstances({})).find((instanceArg) => {
+        instanceArg.data.status === 'bidding' &&
+          instanceArg.data.biddingStartTime <= Date.now() - 4000 &&
+          instanceArg.data.biddingStartTime >= Date.now() - 30000;
+      })
+    ) {
+      console.log('too late to the bidding party... waiting for next round.');
+      return;
     } else {
       await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+        await this.ownInstance.updateFromDb();
         this.ownInstance.data.status = 'bidding';
+        this.ownInstance.data.biddingStartTime = Date.now();
         this.ownInstance.data.biddingShortcode = plugins.smartunique.shortId();
         await this.ownInstance.save();
+        console.log('bidding code stored.');
       });
-      await plugins.smartdelay.delayFor(plugins.smarttime.getMilliSecondsFromUnits({ minutes: 2 }));
+      console.log(`bidding for leadership...`);
+      await plugins.smartdelay.delayFor(
+        plugins.smarttime.getMilliSecondsFromUnits({ seconds: 5 })
+      );
       await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
         let biddingInstances = await DistributedClass.getInstances({});
         biddingInstances = biddingInstances.filter(
           (instanceArg) =>
-            !instanceArg.data.elected &&
+            instanceArg.data.status === 'bidding' &&
             instanceArg.data.lastUpdated >=
-              Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ minutes: 1 })
+              Date.now() - plugins.smarttime.getMilliSecondsFromUnits({ seconds: 10 })
         );
+        console.log(`found ${biddingInstances.length} bidding instances...`);
         this.ownInstance.data.elected = true;
         for (const biddingInstance of biddingInstances) {
           if (biddingInstance.data.biddingShortcode < this.ownInstance.data.biddingShortcode) {
             this.ownInstance.data.elected = false;
           }
         }
+        await plugins.smartdelay.delayFor(5000);
+        console.log(`settling with status elected = ${this.ownInstance.data.elected}`);
+        this.ownInstance.data.status = 'settled';
         await this.ownInstance.save();
       });
       if (this.ownInstance.data.elected) {
@@ -151,20 +213,38 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
    * the leading is implemented here
    */
   public async leadFunction() {
-    const ownInstance = await this.init();
-    const watcher = await DistributedClass.watch({});
-    /**
-     * this function is started once per unique job request
-     */
-    const startResultTimer = async () => {};
+    this.distributedWatcher = await DistributedClass.watch({});
 
-    watcher.changeSubject.subscribe({
+    const currentTaskRequests: Array<{
+      taskName: string;
+      taskExecutionTime: number;
+      /**
+       * all instances that requested this task
+       */
+      requestingDistibutedInstanceIds: string[];
+      responseTimeout: plugins.smartdelay.Timeout<any>;
+    }> = [];
+
+    this.distributedWatcher.changeSubject.subscribe({
       next: async (distributedDoc) => {
+        if (!distributedDoc) {
+          console.log(`registered deletion of instance...`);
+          return;
+        }
+        console.log(distributedDoc);
+        console.log(`registered change for ${distributedDoc.id}`);
         distributedDoc;
       },
     });
-    while (this.ownInstance.data.status !== 'stopped') {
-      await plugins.smartdelay.delayFor(1000);
+
+    while (this.ownInstance.data.status !== 'stopped' && this.ownInstance.data.elected) {
+      const allInstances = await DistributedClass.getInstances({});
+      for (const instance of allInstances) {
+        if (instance.data.status === 'stopped') {
+          await instance.delete();
+        };
+      }
+      await plugins.smartdelay.delayFor(10000);
     }
   }
 
@@ -172,13 +252,29 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
   public async fireDistributedTaskRequest(
     taskRequestArg: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequest
   ): Promise<plugins.taskbuffer.distributedCoordination.IDistributedTaskRequestResult> {
-    const ownInstance = await this.init();
     await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      if (!this.ownInstance) {
+        console.error('instance need to be started first...');
+        return;
+      }
+      await this.ownInstance.updateFromDb();
       this.ownInstance.data.taskRequests.push(taskRequestArg);
       await this.ownInstance.save();
     });
+    await plugins.smartdelay.delayFor(10000);
+    const result = await this.asyncExecutionStack.getExclusiveExecutionSlot(async () => {
+      await this.ownInstance.updateFromDb();
+      const taskRequestResult = this.ownInstance.data.taskRequestResults.find((resultItem) => {
+        return resultItem.requestResponseId === taskRequestArg.requestResponseId;
+      });
+      return taskRequestResult;
+    });
+    if (!result) {
+      console.warn('no result found for task request...');
       return null;
     }
+    return result;
+  }
 
   public async updateDistributedTaskRequest(
     infoBasisArg: plugins.taskbuffer.distributedCoordination.IDistributedTaskRequest
@@ -190,6 +286,10 @@ export class SmartdataDistributedCoordinator extends plugins.taskbuffer.distribu
           infoBasisItem.taskExecutionTime === infoBasisArg.taskExecutionTime
         );
       });
+      if (!existingInfoBasis) {
+        console.warn('trying to update a non existing task request... aborting!');
+        return;
+      }
       Object.assign(existingInfoBasis, infoBasisArg);
       await this.ownInstance.save();
       plugins.smartdelay.delayFor(60000).then(() => {

ts/smartdata.classes.doc.ts

@@ -257,7 +257,7 @@ export class SmartDataDbDoc<T extends TImplements, TImplements, TManager extends
    * updates an object from db
    */
   public async updateFromDb() {
-    const mongoDbNativeDoc = await this.collection.findOne(this.createIdentifiableObject());
+    const mongoDbNativeDoc = await this.collection.findOne(await this.createIdentifiableObject());
     for (const key of Object.keys(mongoDbNativeDoc)) {
       this[key] = mongoDbNativeDoc[key];
     }

ts/smartdata.classes.easystore.ts

@@ -70,7 +70,7 @@ export class EasyStore<T> {
   /**
    * writes a specific key to the keyValueStore
    */
-  public async writeKey(keyArg: keyof T, valueArg: any) {
+  public async writeKey<TKey extends keyof T>(keyArg: TKey, valueArg: T[TKey]) {
     const easyStore = await this.getEasyStore();
     easyStore.data[keyArg] = valueArg;
     await easyStore.save();

ts/smartdata.classes.watcher.ts

@@ -17,9 +17,13 @@ export class SmartdataDbWatcher<T = any> {
     smartdataDbDocArg: typeof SmartDataDbDoc
   ) {
     this.changeStream = changeStreamArg;
-    this.changeStream.on('change', async (item: T) => {
+    this.changeStream.on('change', async (item: any) => {
+      if (!item.fullDocument) {
+        this.changeSubject.next(null);
+        return;
+      }
       this.changeSubject.next(
-        smartdataDbDocArg.createInstanceFromMongoDbNativeDoc(item) as any as T
+        smartdataDbDocArg.createInstanceFromMongoDbNativeDoc(item.fullDocument) as any as T
       );
     });
     plugins.smartdelay.delayFor(0).then(() => {

ts/smartdata.plugins.ts

@@ -8,7 +8,6 @@ import * as lik from '@push.rocks/lik';
 import * as smartdelay from '@push.rocks/smartdelay';
 import * as smartlog from '@push.rocks/smartlog';
 import * as smartpromise from '@push.rocks/smartpromise';
-import * as smartq from '@push.rocks/smartpromise';
 import * as smartrx from '@push.rocks/smartrx';
 import * as smartstring from '@push.rocks/smartstring';
 import * as smarttime from '@push.rocks/smarttime';
@@ -21,7 +20,6 @@ export {
   smartdelay,
   smartpromise,
   smartlog,
-  smartq,
   smartrx,
   mongodb,
   smartstring,