Define a log prefix on child loggers.

Author: rkistner

libs/lib-services/src/logger/Logger.ts

@@ -1,8 +1,22 @@
 import winston from 'winston';
 
+const prefixFormat = winston.format((info) => {
+  if (info.prefix) {
+    info.message = `${info.prefix}${info.message}`;
+  }
+  return {
+    ...info,
+    prefix: undefined
+  };
+});
+
 export namespace LogFormat {
-  export const development = winston.format.combine(winston.format.colorize({ level: true }), winston.format.simple());
-  export const production = winston.format.combine(winston.format.timestamp(), winston.format.json());
+  export const development = winston.format.combine(
+    prefixFormat(),
+    winston.format.colorize({ level: true }),
+    winston.format.simple()
+  );
+  export const production = winston.format.combine(prefixFormat(), winston.format.timestamp(), winston.format.json());
 }
 
 export const logger = winston.createLogger();

modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts

@@ -7,7 +7,8 @@ import {
   container,
   ErrorCode,
   errors,
-  logger,
+  Logger,
+  logger as defaultLogger,
   ReplicationAssertionError,
   ServiceError
 } from '@powersync/lib-services-framework';
@@ -55,12 +56,16 @@ export interface MongoBucketBatchOptions {
   skipExistingRows: boolean;
 
   markRecordUnavailable: BucketStorageMarkRecordUnavailable | undefined;
+
+  logger?: Logger;
 }
 
 export class MongoBucketBatch
   extends BaseObserver<storage.BucketBatchStorageListener>
   implements storage.BucketStorageBatch
 {
+  private logger: Logger;
+
   private readonly client: mongo.MongoClient;
   public readonly db: PowerSyncMongo;
   public readonly session: mongo.ClientSession;
@@ -96,6 +101,7 @@ export class MongoBucketBatch
 
   constructor(options: MongoBucketBatchOptions) {
     super();
+    this.logger = options.logger ?? defaultLogger;
     this.client = options.db.client;
     this.db = options.db;
     this.group_id = options.groupId;
@@ -242,7 +248,9 @@ export class MongoBucketBatch
         current_data_lookup.set(cacheKey(doc._id.t, doc._id.k), doc);
       }
 
-      let persistedBatch: PersistedBatch | null = new PersistedBatch(this.group_id, transactionSize);
+      let persistedBatch: PersistedBatch | null = new PersistedBatch(this.group_id, transactionSize, {
+        logger: this.logger
+      });
 
       for (let op of b) {
         if (resumeBatch) {
@@ -329,7 +337,7 @@ export class MongoBucketBatch
             this.markRecordUnavailable(record);
           } else {
             // Log to help with debugging if there was a consistency issue
-            logger.warn(
+            this.logger.warn(
               `Cannot find previous record for update on ${record.sourceTable.qualifiedName}: ${beforeId} / ${record.before?.id}`
             );
           }
@@ -350,7 +358,7 @@ export class MongoBucketBatch
         existing_lookups = [];
         // Log to help with debugging if there was a consistency issue
         if (this.storeCurrentData) {
-          logger.warn(
+          this.logger.warn(
             `Cannot find previous record for delete on ${record.sourceTable.qualifiedName}: ${beforeId} / ${record.before?.id}`
           );
         }
@@ -447,7 +455,7 @@ export class MongoBucketBatch
               }
             }
           );
-          logger.error(
+          this.logger.error(
             `Failed to evaluate data query on ${record.sourceTable.qualifiedName}.${record.after?.id}: ${error.error}`
           );
         }
@@ -487,7 +495,7 @@ export class MongoBucketBatch
               }
             }
           );
-          logger.error(
+          this.logger.error(
             `Failed to evaluate parameter query on ${record.sourceTable.qualifiedName}.${after.id}: ${error.error}`
           );
         }
@@ -541,7 +549,7 @@ export class MongoBucketBatch
             if (e instanceof mongo.MongoError && e.hasErrorLabel('TransientTransactionError')) {
               // Likely write conflict caused by concurrent write stream replicating
             } else {
-              logger.warn('Transaction error', e as Error);
+              this.logger.warn('Transaction error', e as Error);
             }
             await timers.setTimeout(Math.random() * 50);
             throw e;
@@ -566,7 +574,7 @@ export class MongoBucketBatch
     await this.withTransaction(async () => {
       flushTry += 1;
       if (flushTry % 10 == 0) {
-        logger.info(`${this.slot_name} ${description} - try ${flushTry}`);
+        this.logger.info(`${description} - try ${flushTry}`);
       }
       if (flushTry > 20 && Date.now() > lastTry) {
         throw new ServiceError(ErrorCode.PSYNC_S1402, 'Max transaction tries exceeded');
@@ -636,12 +644,12 @@ export class MongoBucketBatch
     if (this.last_checkpoint_lsn != null && lsn < this.last_checkpoint_lsn) {
       // When re-applying transactions, don't create a new checkpoint until
       // we are past the last transaction.
-      logger.info(`Re-applied transaction ${lsn} - skipping checkpoint`);
+      this.logger.info(`Re-applied transaction ${lsn} - skipping checkpoint`);
       return false;
     }
     if (lsn < this.no_checkpoint_before_lsn) {
       if (Date.now() - this.lastWaitingLogThottled > 5_000) {
-        logger.info(
+        this.logger.info(
           `Waiting until ${this.no_checkpoint_before_lsn} before creating checkpoint, currently at ${lsn}. Persisted op: ${this.persisted_op}`
         );
         this.lastWaitingLogThottled = Date.now();
@@ -713,7 +721,7 @@ export class MongoBucketBatch
     if (this.persisted_op != null) {
       // The commit may have been skipped due to "no_checkpoint_before_lsn".
       // Apply it now if relevant
-      logger.info(`Commit due to keepalive at ${lsn} / ${this.persisted_op}`);
+      this.logger.info(`Commit due to keepalive at ${lsn} / ${this.persisted_op}`);
       return await this.commit(lsn);
     }
 
@@ -777,7 +785,7 @@ export class MongoBucketBatch
       return null;
     }
 
-    logger.debug(`Saving ${record.tag}:${record.before?.id}/${record.after?.id}`);
+    this.logger.debug(`Saving ${record.tag}:${record.before?.id}/${record.after?.id}`);
 
     this.batch ??= new OperationBatch();
     this.batch.push(new RecordOperation(record));
@@ -848,7 +856,7 @@ export class MongoBucketBatch
           session: session
         });
         const batch = await cursor.toArray();
-        const persistedBatch = new PersistedBatch(this.group_id, 0);
+        const persistedBatch = new PersistedBatch(this.group_id, 0, { logger: this.logger });
 
         for (let value of batch) {
           persistedBatch.saveBucketData({

modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts

@@ -138,6 +138,7 @@ export class MongoSyncBucketStorage
     const checkpoint_lsn = doc?.last_checkpoint_lsn ?? null;
 
     await using batch = new MongoBucketBatch({
+      logger: options.logger,
       db: this.db,
       syncRules: this.sync_rules.parsed(options).sync_rules,
       groupId: this.group_id,

modules/module-mongodb-storage/src/storage/implementation/PersistedBatch.ts

@@ -3,7 +3,7 @@ import { JSONBig } from '@powersync/service-jsonbig';
 import { EvaluatedParameters, EvaluatedRow } from '@powersync/service-sync-rules';
 import * as bson from 'bson';
 
-import { logger } from '@powersync/lib-services-framework';
+import { Logger, logger as defaultLogger } from '@powersync/lib-services-framework';
 import { InternalOpId, storage, utils } from '@powersync/service-core';
 import { currentBucketKey, MAX_ROW_SIZE } from './MongoBucketBatch.js';
 import { MongoIdSequence } from './MongoIdSequence.js';
@@ -46,6 +46,7 @@ const MAX_TRANSACTION_DOC_COUNT = 2_000;
  * multiple transactions.
  */
 export class PersistedBatch {
+  logger: Logger;
   bucketData: mongo.AnyBulkWriteOperation<BucketDataDocument>[] = [];
   bucketParameters: mongo.AnyBulkWriteOperation<BucketParameterDocument>[] = [];
   currentData: mongo.AnyBulkWriteOperation<CurrentDataDocument>[] = [];
@@ -63,9 +64,11 @@ export class PersistedBatch {
 
   constructor(
     private group_id: number,
-    writtenSize: number
+    writtenSize: number,
+    options?: { logger?: Logger }
   ) {
     this.currentSize = writtenSize;
+    this.logger = options?.logger ?? defaultLogger;
   }
 
   private incrementBucket(bucket: string, op_id: InternalOpId) {
@@ -107,7 +110,7 @@ export class PersistedBatch {
         // the BSON size is small enough, but the JSON size is too large.
         // In these cases, we can't store the data, so we skip it, or generate a REMOVE operation if the row
         // was synced previously.
-        logger.error(`powersync_${this.group_id} Row ${key} too large: ${recordData.length} bytes. Removing.`);
+        this.logger.error(`Row ${key} too large: ${recordData.length} bytes. Removing.`);
         continue;
       }
 
@@ -303,8 +306,8 @@ export class PersistedBatch {
     }
 
     const duration = performance.now() - startAt;
-    logger.info(
-      `powersync_${this.group_id} Flushed ${this.bucketData.length} + ${this.bucketParameters.length} + ${
+    this.logger.info(
+      `Flushed ${this.bucketData.length} + ${this.bucketParameters.length} + ${
         this.currentData.length
       } updates, ${Math.round(this.currentSize / 1024)}kb in ${duration.toFixed(0)}ms. Last op_id: ${this.debugLastOpId}`
     );