feat: tile rendering pre-process cooldown validation (#39)

* feat: tile processing skip due to cooldown

* feat: cooldown functionality improvements

* test: aws-sdk requires globalSetup fix

* fix: metric name typo

Author: melancholiai

README.md

@@ -18,13 +18,22 @@ The fetched metatile image will be splitted into 256x256 pixels tiles in a PNG f
 ## How it works
 ```mermaid
 flowchart TD
-    A[Start] --> B{IsQueueEmpty}
+    A[Start] --> B{Is Queue Empty?}
     B -- yes --> G[Finish]
     B -- no --> C([Fetch Tile])
-    C -->D([Get Map])
+    C --> I{Is Forced?}
+    I -- no -->J([Get Tile Details])
+    J -->N([Get Project Details])
+    N -->M{Is Tile Up To Date?}
+    M -- yes, skipped -->H
+    M -- no -->O{Is Cooledowned?}
+    O -- yes, cooled --> H
+    O -- no --> D
+    I -- yes -->D([Get Map])
     D -->E([Split Map])
     E -->F([Store Tiles])
-    F --> B
+    F -- rendered -->H([Upsert Tile Details])
+    H --> B
 ```
 
 ## config

package-lock.json

@@ -45,8 +45,8 @@
     "@aws-sdk/client-s3": "^3.53.1",
     "@godaddy/terminus": "4.9.0",
     "@map-colonies/cleanup-registry": "^1.1.0",
-    "@map-colonies/detiler-client": "^1.0.0",
-    "@map-colonies/detiler-common": "^1.0.0",
+    "@map-colonies/detiler-client": "^1.3.0-rc2",
+    "@map-colonies/detiler-common": "^1.3.0-rc2",
     "@map-colonies/error-express-handler": "^2.1.0",
     "@map-colonies/express-access-log-middleware": "^1.0.0",
     "@map-colonies/js-logger": "^0.0.5",

package.json

@@ -3,6 +3,7 @@ import { Logger } from '@map-colonies/js-logger';
 import { IDetilerClient } from '@map-colonies/detiler-client';
 import { inject, injectable } from 'tsyringe';
 import { AxiosInstance } from 'axios';
+import { TILEGRID_WORLD_CRS84, tileToBoundingBox } from '@map-colonies/tile-calc';
 import { IConfig } from '../common/interfaces';
 import { IProjectConfig } from '../common/interfaces';
 import { fetchTimestampValue, timestampToUnix } from '../common/util';
@@ -18,13 +19,22 @@ import {
 import { MapProvider, MapSplitterProvider, TilesStorageProvider } from './interfaces';
 import { TileWithMetadata } from './types';
 
+type SkipReason = 'tile_up_to_date' | 'cooldown';
+type ProcessReason = 'project_updated' | 'force' | 'no_detiler' | 'error_occurred';
+
+interface PreProcessReult {
+  shouldSkipProcessing: boolean;
+  reason?: ProcessReason | SkipReason;
+}
+
 @injectable()
 export class TileProcessor {
   private readonly project: IProjectConfig;
   private readonly forceProcess: boolean;
   private readonly detilerProceedOnFailure: boolean;
 
   private readonly tilesCounter?: client.Counter<'status' | 'z'>;
+  private readonly preProcessResultsCounter?: client.Counter<'result' | 'z'>;
   private readonly tilesDurationHistogram?: client.Histogram<'z' | 'kind'>;
 
   public constructor(
@@ -57,6 +67,13 @@ export class TileProcessor {
         labelNames: ['status', 'z'] as const,
         registers: [registry],
       });
+
+      this.preProcessResultsCounter = new client.Counter({
+        name: 'retiler_pre_process_results_count',
+        help: 'The results of the pre process',
+        labelNames: ['result', 'z'] as const,
+        registers: [registry],
+      });
     }
   }
 
@@ -66,8 +83,9 @@ export class TileProcessor {
       const preRenderTimestamp = Math.floor(Date.now() / MILLISECONDS_IN_SECOND);
 
       // check if possibly the tile processing can be skipped according to detiler
-      const shouldSkip = await this.preProcess(tile, preRenderTimestamp);
-      if (shouldSkip) {
+      const { shouldSkipProcessing } = await this.preProcess(tile, preRenderTimestamp);
+
+      if (shouldSkipProcessing) {
         this.tilesCounter?.inc({ status: 'skipped', z: tile.z });
         return;
       }
@@ -104,16 +122,21 @@ export class TileProcessor {
     }
   }
 
-  private async preProcess(tile: TileWithMetadata, timestamp: number): Promise<boolean> {
-    const isForced = this.forceProcess || tile.force === true;
+  private async preProcess(tile: TileWithMetadata, timestamp: number): Promise<PreProcessReult> {
+    let preProcessTimerEnd;
+    let result: PreProcessReult = { shouldSkipProcessing: false };
 
-    if (this.detiler === undefined || isForced) {
-      return false;
-    }
+    try {
+      // check for forced rendering or if detiler option is off
+      const isForced = this.forceProcess || tile.force === true;
 
-    const detilerGetTimerEnd = this.tilesDurationHistogram?.startTimer({ kind: 'detilerGet' });
+      if (isForced || this.detiler === undefined) {
+        result = { shouldSkipProcessing: false, reason: isForced ? 'force' : 'no_detiler' };
+        return result;
+      }
+
+      preProcessTimerEnd = this.tilesDurationHistogram?.startTimer({ kind: 'pre_process' });
 
-    try {
       // attempt to get latest tile details
       const tileDetails = await this.detiler.getTileDetails({ kit: this.project.name, z: tile.z, x: tile.x, y: tile.y });
 
@@ -129,20 +152,77 @@ export class TileProcessor {
         if (tileDetails.renderedAt >= projectTimestamp) {
           await this.detiler.setTileDetails(
             { kit: this.project.name, z: tile.z, x: tile.x, y: tile.y },
-            { hasSkipped: true, state: tile.state, timestamp }
+            { status: 'skipped', state: tile.state, timestamp }
           );
-          this.logger.info({ msg: 'skipping tile processing', tile, tileDetails, sourceUpdatedAt: projectTimestamp });
-          return true;
+
+          this.logger.info({
+            msg: 'tile processing can be skipping due to tile being up do date',
+            tile,
+            tileDetails,
+            sourceUpdatedAt: projectTimestamp,
+          });
+
+          result = { shouldSkipProcessing: true, reason: 'tile_up_to_date' };
+
+          return result;
+        }
+
+        // tile geometry in bbox
+        const { west, south, east, north } = tileToBoundingBox(tile, TILEGRID_WORLD_CRS84, true);
+
+        // time elapsed since last rendered
+        const cooled = timestamp - tileDetails.renderedAt;
+
+        // only render if the time elapsed is longer than the relavant cooldowns duration otherwise the tile is still cooling
+        const cooldownsGenerator = this.detiler.queryCooldownsAsyncGenerator({
+          enabled: true,
+          minZoom: tile.z,
+          maxZoom: tile.z,
+          kits: [this.project.name],
+          area: [west, south, east, north],
+        });
+
+        for await (const cooldowns of cooldownsGenerator) {
+          const isCooling = cooldowns.filter((cooldown) => cooldown.duration > cooled).length > 0;
+
+          this.logger.info({
+            msg: 'tile processing should be skipped due to active cooldown',
+            tile,
+            tileDetails,
+            tileCooled: cooled,
+            cooldowns,
+            sourceUpdatedAt: projectTimestamp,
+          });
+
+          if (isCooling) {
+            await this.detiler.setTileDetails(
+              { kit: this.project.name, z: tile.z, x: tile.x, y: tile.y },
+              { status: 'cooled', state: tile.state, timestamp }
+            );
+
+            result = { shouldSkipProcessing: true, reason: 'cooldown' };
+
+            return result;
+          }
         }
       }
 
-      return false;
+      result = { shouldSkipProcessing: false, reason: 'project_updated' };
+
+      return result;
     } catch (error) {
       this.logger.error({ msg: 'an error occurred while pre processing, tile will be processed', error });
-      return false;
+
+      result = { shouldSkipProcessing: false, reason: 'error_occurred' };
+
+      return result;
     } finally {
-      if (detilerGetTimerEnd) {
-        detilerGetTimerEnd();
+      this.logger.info({ msg: 'pre processing done', tile, result });
+
+      this.preProcessResultsCounter?.inc({ result: result.reason, z: tile.z });
+
+      if (preProcessTimerEnd) {
+        preProcessTimerEnd();
       }
     }
   }
@@ -152,18 +232,21 @@ export class TileProcessor {
       return;
     }
 
-    const detilerSetTimerEnd = this.tilesDurationHistogram?.startTimer({ kind: 'detilerSet' });
+    const postProcessTimerEnd = this.tilesDurationHistogram?.startTimer({ kind: 'post_process' });
 
     try {
-      await this.detiler.setTileDetails({ kit: this.project.name, z: tile.z, x: tile.x, y: tile.y }, { state: tile.state, timestamp });
+      await this.detiler.setTileDetails(
+        { kit: this.project.name, z: tile.z, x: tile.x, y: tile.y },
+        { status: 'rendered', state: tile.state, timestamp }
+      );
     } catch (error) {
       this.logger.error({ msg: 'an error occurred while post processing, skipping details set', error });
       if (!this.detilerProceedOnFailure) {
         throw error;
       }
     } finally {
-      if (detilerSetTimerEnd) {
-        detilerSetTimerEnd();
+      if (postProcessTimerEnd) {
+        postProcessTimerEnd();
       }
     }
   }

src/retiler/tileProcessor.ts

@@ -3,6 +3,8 @@ import { S3Client, CreateBucketCommand, HeadBucketCommand } from '@aws-sdk/clien
 import config from 'config';
 import { S3StorageProviderConfig, StorageProviderConfig } from '../../../src/retiler/tilesStorageProvider/interfaces';
 
+process.env.ALLOW_CONFIG_MUTATIONS = 'true'; // @aws-sdk/client-s3 attempts to modify config on tests
+
 export default async (): Promise<void> => {
   const storageProvidersConfig = config.get<StorageProviderConfig[]>('app.tilesStorage.providers');
 

tests/configurations/integration/jest.globalSetup.ts

@@ -58,6 +58,7 @@ describe('retiler', function () {
   let stateInterceptor: nock.Interceptor;
   let detilerScope: nock.Scope;
   let detilerGetInterceptor: nock.Interceptor;
+  let cooldownsGetInterceptor: nock.Interceptor;
   let detilerPutInterceptor: nock.Interceptor;
   let stateBuffer: Buffer;
   let mapBuffer2048x2048: Buffer;
@@ -72,7 +73,8 @@ describe('retiler', function () {
     getMapInterceptor = nock(mapUrl).defaultReplyHeaders({ 'content-type': 'image/png' }).get(/.*/);
     stateInterceptor = nock(stateUrl).get(/.*/);
     detilerScope = nock(detilerUrl);
-    detilerGetInterceptor = detilerScope.get(/.*/);
+    detilerGetInterceptor = detilerScope.get(/^\/detail/);
+    cooldownsGetInterceptor = detilerScope.get(/^\/cooldown/);
     detilerPutInterceptor = detilerScope.put(/.*/);
     stateBuffer = await fsPromises.readFile('tests/state.txt');
     mapBuffer512x512 = await fsPromises.readFile('tests/512x512.png');
@@ -83,6 +85,7 @@ describe('retiler', function () {
     nock.removeInterceptor(getMapInterceptor);
     nock.removeInterceptor(stateInterceptor);
     nock.removeInterceptor(detilerGetInterceptor);
+    nock.removeInterceptor(cooldownsGetInterceptor);
     nock.removeInterceptor(detilerPutInterceptor);
     jest.clearAllMocks();
   });
@@ -297,6 +300,83 @@ describe('retiler', function () {
         LONG_RUNNING_TEST
       );
 
+      it(
+        'should complete a single job where tile is not skipped even if a cooldown is found',
+        async function () {
+          detilerGetInterceptor.reply(httpStatusCodes.OK, { renderedAt: 0 });
+          cooldownsGetInterceptor.reply(httpStatusCodes.OK, [{ duration: 1 }]);
+          detilerPutInterceptor.reply(httpStatusCodes.OK);
+          const stateScope = stateInterceptor.reply(httpStatusCodes.OK, stateBuffer);
+          const getMapScope = getMapInterceptor.reply(httpStatusCodes.OK, mapBuffer512x512);
+
+          const pgBoss = container.resolve(PgBoss);
+          const provider = container.resolve<PgBossJobQueueProvider>(JOB_QUEUE_PROVIDER);
+          const queueName = container.resolve<string>(QUEUE_NAME);
+          const jobId = await pgBoss.send({ name: queueName, data: { z: 1, x: 0, y: 0, metatile: 2, parent: 'parent' } });
+
+          const consumePromise = consumeAndProcessFactory(container)();
+
+          const storageProviders = container.resolve<TilesStorageProvider[]>(TILES_STORAGE_PROVIDERS);
+          const storeTileSpies = storageProviders.map((provider) => jest.spyOn(provider, 'storeTile'));
+
+          const job = await waitForJobToBeResolved(pgBoss, jobId as string);
+          await provider.stopQueue();
+
+          await expect(consumePromise).resolves.not.toThrow();
+
+          expect(job).toHaveProperty('state', 'completed');
+
+          storeTileSpies.forEach((spy) => expect(spy.mock.calls).toHaveLength(4));
+
+          for (const storeTileSpy of storeTileSpies) {
+            for (let i = 0; i < 4; i++) {
+              const storeCall = storeTileSpy.mock.calls[i][0];
+              const key = determineKey({ x: storeCall.x, y: storeCall.y, z: storeCall.z, metatile: storeCall.metatile });
+              const expectedBuffer = await fsPromises.readFile(`tests/integration/expected/${key}`);
+              expect(expectedBuffer.compare(storeCall.buffer)).toBe(0);
+            }
+          }
+
+          getMapScope.done();
+          detilerScope.done();
+          stateScope.done();
+        },
+        LONG_RUNNING_TEST
+      );
+
+      it(
+        'should complete a single job where tile processing is skipped due to cooldown',
+        async function () {
+          detilerGetInterceptor.reply(httpStatusCodes.OK, { renderedAt: 0 });
+          cooldownsGetInterceptor.reply(httpStatusCodes.OK, [{ duration: 9999999999 }]);
+          detilerPutInterceptor.reply(httpStatusCodes.OK);
+          const stateScope = stateInterceptor.reply(httpStatusCodes.OK, stateBuffer);
+
+          const pgBoss = container.resolve(PgBoss);
+          const provider = container.resolve<PgBossJobQueueProvider>(JOB_QUEUE_PROVIDER);
+          const queueName = container.resolve<string>(QUEUE_NAME);
+          const jobId = await pgBoss.send({ name: queueName, data: { z: 1, x: 0, y: 0, metatile: 2, parent: 'parent' } });
+
+          const consumePromise = consumeAndProcessFactory(container)();
+
+          const storageProviders = container.resolve<TilesStorageProvider[]>(TILES_STORAGE_PROVIDERS);
+          const storeTileSpies = storageProviders.map((provider) => jest.spyOn(provider, 'storeTile'));
+
+          const job = await waitForJobToBeResolved(pgBoss, jobId as string);
+          await provider.stopQueue();
+
+          await expect(consumePromise).resolves.not.toThrow();
+
+          expect(job).toHaveProperty('state', 'completed');
+
+          storeTileSpies.forEach((spy) => expect(spy.mock.calls).toHaveLength(0));
+
+          detilerScope.done();
+          stateScope.done();
+        },
+        LONG_RUNNING_TEST
+      );
+
       it(
         'should complete a single job where tile is forced',
         async function () {