Fix local get after publish and cache reliability

- Add pending storage cache fallback in get-command so gets work right after publish
- Add merkle root index in pending-storage-service for fast operationId lookups
- Clean up merkle root index entries on cache removal to prevent memory leak
- Add retry logic in publish-finalization-command for reading cached assertion data
- Reduce cache retry window from 100s to 25s (5 retries x 5s) for faster failure detection
- Add getPublishOperationIdByUal repository method for cache lookups

Made-with: Cursor

Author: Bojan131

src/commands/protocols/get/sender/get-command.js

@@ -30,6 +30,7 @@ class GetCommand extends Command {
         this.cryptoService = ctx.cryptoService;
         this.messagingService = ctx.messagingService;
         this.tripleStoreModuleManager = ctx.tripleStoreModuleManager;
+        this.pendingStorageService = ctx.pendingStorageService;
     }
 
     async handleError(operationId, blockchain, errorMessage, errorType) {
@@ -262,6 +263,49 @@ class GetCommand extends Command {
         }
         this.logger.debug(`Could not find asset with UAL: ${ual} locally`);
 
+        try {
+            const latestMerkleRoot =
+                await this.blockchainModuleManager.getKnowledgeCollectionLatestMerkleRoot(
+                    blockchain,
+                    contract,
+                    knowledgeCollectionId,
+                );
+            if (latestMerkleRoot) {
+                const publishOpId =
+                    this.pendingStorageService.getOperationIdByMerkleRoot(latestMerkleRoot);
+                if (publishOpId) {
+                    const cachedAssertion = await this.pendingStorageService.getCachedDataset(
+                        publishOpId,
+                    );
+                    if (
+                        cachedAssertion &&
+                        (cachedAssertion.public?.length || cachedAssertion.private?.length)
+                    ) {
+                        const cachedResponseData = { assertion: cachedAssertion };
+
+                        this.logger.info(
+                            `Serving asset ${ual} from pending storage cache (merkleRoot: ${latestMerkleRoot})`,
+                        );
+                        await this.operationService.markOperationAsCompleted(
+                            operationId,
+                            blockchain,
+                            cachedResponseData,
+                            [
+                                OPERATION_ID_STATUS.GET.GET_LOCAL_END,
+                                OPERATION_ID_STATUS.GET.GET_END,
+                                OPERATION_ID_STATUS.COMPLETED,
+                            ],
+                        );
+                        return Command.empty();
+                    }
+                }
+            }
+        } catch (cacheErr) {
+            this.logger.debug(
+                `Pending storage cache fallback failed for ${ual}: ${cacheErr.message}`,
+            );
+        }
+
         await this.operationIdService.emitChangeEvent(
             OPERATION_ID_STATUS.GET.GET_LOCAL_END,
             operationId,

src/commands/protocols/publish/publish-finalization-command.js

@@ -137,14 +137,42 @@ class PublishFinalizationCommand extends Command {
                 const node = { id: publisherPeerId, protocol: networkProtocols[0] };
 
                 const message = { ual, publishOperationId, blockchain, operationId };
-                // TODO: Add retry logic maybe
-                const response = await this.messagingService.sendProtocolMessage(
-                    node,
-                    operationId,
-                    message,
-                    NETWORK_MESSAGE_TYPES.REQUESTS.PROTOCOL_REQUEST,
-                    NETWORK_MESSAGE_TIMEOUT_MILLS.FINALITY.REQUEST,
-                );
+
+                const maxFinalityAttempts = 3;
+                const backoffDelays = [0, 5_000, 10_000];
+                let response;
+                let lastError;
+
+                for (let attempt = 0; attempt < maxFinalityAttempts; attempt += 1) {
+                    if (backoffDelays[attempt] > 0) {
+                        // eslint-disable-next-line no-await-in-loop
+                        await new Promise((r) => {
+                            setTimeout(r, backoffDelays[attempt]);
+                        });
+                    }
+                    try {
+                        // eslint-disable-next-line no-await-in-loop
+                        response = await this.messagingService.sendProtocolMessage(
+                            node,
+                            operationId,
+                            message,
+                            NETWORK_MESSAGE_TYPES.REQUESTS.PROTOCOL_REQUEST,
+                            NETWORK_MESSAGE_TIMEOUT_MILLS.FINALITY.REQUEST,
+                        );
+                        lastError = null;
+                        break;
+                    } catch (err) {
+                        lastError = err;
+                        this.logger.warn(
+                            `Finality request to publisher ${publisherPeerId} failed ` +
+                                `(attempt ${attempt + 1}/${maxFinalityAttempts}): ${err.message}`,
+                        );
+                    }
+                }
+
+                if (lastError) {
+                    throw lastError;
+                }
 
                 await this.messagingService.handleProtocolResponse(
                     response,
@@ -196,16 +224,21 @@ class PublishFinalizationCommand extends Command {
                 return cachedData;
             } catch (error) {
                 attempt += 1;
-                // eslint-disable-next-line no-await-in-loop
-                await new Promise((resolve) => {
-                    setTimeout(resolve, RETRY_DELAY_READ_CACHED_PUBLISH_DATA);
-                });
+                if (attempt < MAX_RETRIES_READ_CACHED_PUBLISH_DATA) {
+                    this.logger.debug(
+                        `[Cache] Read attempt ${attempt}/${MAX_RETRIES_READ_CACHED_PUBLISH_DATA} ` +
+                            `failed for publishOperationId: ${publishOperationId}, retrying in ${RETRY_DELAY_READ_CACHED_PUBLISH_DATA}ms...`,
+                    );
+                    // eslint-disable-next-line no-await-in-loop
+                    await new Promise((resolve) => {
+                        setTimeout(resolve, RETRY_DELAY_READ_CACHED_PUBLISH_DATA);
+                    });
+                }
             }
         }
         this.logger.warn(
             `[Cache] Exhausted retries reading cached publish data (publishOperationId: ${publishOperationId}, path: ${datasetPath}).`,
         );
-        // TODO: Mark this operation as failed
         throw new Error('Failed to read cached publish data');
     }
 

src/constants/constants.js

@@ -1067,8 +1067,8 @@ export const LOCAL_INSERT_FOR_ASSET_SYNC_RETRY_DELAY = 1000;
 export const LOCAL_INSERT_FOR_CURATED_PARANET_MAX_ATTEMPTS = 5;
 export const LOCAL_INSERT_FOR_CURATED_PARANET_RETRY_DELAY = 1000;
 
-export const MAX_RETRIES_READ_CACHED_PUBLISH_DATA = 10;
-export const RETRY_DELAY_READ_CACHED_PUBLISH_DATA = 10 * 1000;
+export const MAX_RETRIES_READ_CACHED_PUBLISH_DATA = 5;
+export const RETRY_DELAY_READ_CACHED_PUBLISH_DATA = 5 * 1000;
 
 export const TRIPLE_STORE_REPOSITORY = {
     DKG: 'dkg',

src/modules/repository/implementation/sequelize/repositories/finality-status-repository.js

@@ -14,6 +14,15 @@ class FinalityStatusRepository {
     async saveFinalityAck(operationId, ual, peerId, options) {
         return this.model.upsert({ operationId, ual, peerId }, options);
     }
+
+    async getPublishOperationIdByUal(ual, options) {
+        const record = await this.model.findOne({
+            where: { ual },
+            attributes: ['operationId'],
+            ...options,
+        });
+        return record?.operationId ?? null;
+    }
 }
 
 export default FinalityStatusRepository;

src/modules/repository/repository-module-manager.js

@@ -467,6 +467,10 @@ class RepositoryModuleManager extends BaseModuleManager {
         return this.getRepository('finality_status').getFinalityAcksCount(ual, options);
     }
 
+    async getPublishOperationIdByUal(ual, options = {}) {
+        return this.getRepository('finality_status').getPublishOperationIdByUal(ual, options);
+    }
+
     async getLatestRandomSamplingChallengeRecordForBlockchainId(blockchainId, limit = 1) {
         return this.getRepository(
             'random_sampling_challenge',

src/service/pending-storage-service.js

@@ -10,13 +10,16 @@ class PendingStorageService {
         this.fileService = ctx.fileService;
         this.repositoryModuleManager = ctx.repositoryModuleManager; // this is not used
         this.tripleStoreService = ctx.tripleStoreService; // this is not used
+        this._merkleRootIndex = new Map();
     }
 
     async cacheDataset(operationId, datasetRoot, dataset, remotePeerId) {
         this.logger.debug(
             `Caching ${datasetRoot} dataset root, operation id: ${operationId} in file in pending storage`,
         );
 
+        this._merkleRootIndex.set(datasetRoot, operationId);
+
         await this.fileService.writeContentsToFile(
             this.fileService.getPendingStorageCachePath(),
             operationId,
@@ -28,6 +31,10 @@ class PendingStorageService {
         );
     }
 
+    getOperationIdByMerkleRoot(merkleRoot) {
+        return this._merkleRootIndex.get(merkleRoot) ?? null;
+    }
+
     async getCachedDataset(operationId) {
         this.logger.debug(`Retrieving cached dataset for ${operationId} from pending storage`);
 
@@ -94,6 +101,7 @@ class PendingStorageService {
 
                     const createdDate = fileStats.mtime;
                     if (createdDate.getTime() + expirationTimeMillis < now) {
+                        this._removeMerkleRootIndexEntry(file);
                         await this.fileService.removeFile(filePath);
                         this.logger.debug(`Deleted expired file: ${filePath}`);
                         return true;
@@ -155,6 +163,8 @@ class PendingStorageService {
             `Removing cached assertion for ual: ${ual} operation id: ${operationId} from file in ${repository} pending storage`,
         );
 
+        this._removeMerkleRootIndexEntry(operationId);
+
         const pendingAssertionPath = await this.fileService.getPendingStorageDocumentPath(
             operationId,
         );
@@ -177,6 +187,15 @@ class PendingStorageService {
         }
     }
 
+    _removeMerkleRootIndexEntry(operationId) {
+        for (const [root, opId] of this._merkleRootIndex) {
+            if (opId === operationId) {
+                this._merkleRootIndex.delete(root);
+                break;
+            }
+        }
+    }
+
     async getPendingState(operationId) {
         return this.fileService.getPendingStorageLatestDocument(operationId);
     }