Merge pull request #1172 from golemfactory/fix/expire-old-proposals

Expire old proposals to prevent memory consumption over time

Author: SewerynKras

src/golem-network/golem-network.ts

@@ -37,6 +37,7 @@ import { IProposalRepository } from "../market/proposal";
 import { Subscription } from "rxjs";
 import { GolemConfigError, GolemUserError } from "../shared/error/golem-error";
 import { GolemPluginInitializer, GolemPluginOptions, GolemPluginRegistration } from "./plugin";
+import { ExpirationManager } from "../shared/expiration/ExpirationManager";
 
 /**
  * Instance of an object or a factory function that you can call `new` on.
@@ -68,6 +69,21 @@ function getFactory<
   return (...args) => new defaultFactory(...args);
 }
 
+interface MarketProposalExpirationOptions {
+  /**
+   * Number of milliseconds before a market proposal is considered stale and will be
+   * removed from all internal caches and DraftOfferProposalPools created by `oneOf` and `manyOf`.
+   * If unspecified it defaults to twice the demand refresh interval (`market.demandRefreshIntervalSec`) if
+   * that's specified or 30 minutes otherwise.
+   */
+  offerProposalTTLMs: number;
+
+  /**
+   * Interval at which to check for expired proposals in milliseconds. Defaults to 60s.
+   */
+  offerProposalCleanupIntervalMs: number;
+}
+
 export interface GolemNetworkOptions {
   /**
    * Logger instance to use for logging.
@@ -98,7 +114,7 @@ export interface GolemNetworkOptions {
    * This is where you can globally specify several options that determine how the SDK will
    * interact with the market.
    */
-  market?: Partial<MarketModuleOptions>;
+  market?: Partial<MarketModuleOptions> & Partial<MarketProposalExpirationOptions>;
 
   /**
    * Set the data transfer protocol to use for file transfers.
@@ -197,6 +213,7 @@ export type GolemServices = {
   demandRepository: IDemandRepository;
   fileServer: IFileServer;
   storageProvider: StorageProvider;
+  proposalExpirationManager: ExpirationManager;
 };
 
 /**
@@ -267,8 +284,23 @@ export class GolemNetwork {
       const demandCache = new CacheService<Demand>();
       const proposalCache = new CacheService<OfferProposal>();
 
+      const proposalExpirationManager = new ExpirationManager({
+        logger: this.logger.child("expiration-manager"),
+        intervalMs: options.market?.offerProposalCleanupIntervalMs || 60 * 1000, // default 60s
+        timeToLiveMs:
+          options.market?.offerProposalTTLMs ||
+          (options?.market?.demandRefreshIntervalSec
+            ? options?.market?.demandRefreshIntervalSec * 1000 * 2
+            : 30 * 60 * 1000),
+      });
+
       const demandRepository = new DemandRepository(this.yagna.market, demandCache);
-      const proposalRepository = new ProposalRepository(this.yagna.market, this.yagna.identity, proposalCache);
+      const proposalRepository = new ProposalRepository(
+        this.yagna.market,
+        this.yagna.identity,
+        proposalCache,
+        proposalExpirationManager,
+      );
       const agreementRepository = new AgreementRepository(this.yagna.market, demandRepository);
 
       this.services = {
@@ -278,6 +310,7 @@ export class GolemNetwork {
         demandRepository,
         proposalCache,
         proposalRepository,
+        proposalExpirationManager,
         paymentApi:
           this.options.override?.paymentApi ||
           new PaymentApiAdapter(
@@ -338,6 +371,7 @@ export class GolemNetwork {
       await this.services.paymentApi.connect();
       await this.storageProvider.init();
       await this.connectPlugins();
+      this.services.proposalExpirationManager.start();
       this.events.emit("connected");
       this.hasConnection = true;
     } catch (err) {
@@ -359,6 +393,7 @@ export class GolemNetwork {
         .catch((err) =>
           this.logger.warn("Closing connections with yagna resulted with an error, it will be ignored", err),
         );
+      this.services.proposalExpirationManager.stopAndReset();
       this.services.proposalCache.flushAll();
       this.abortController = new AbortController();
     } catch (err) {
@@ -487,6 +522,7 @@ export class GolemNetwork {
         logger: this.logger,
         validateOfferProposal: order.market.offerProposalFilter,
         selectOfferProposal: order.market.offerProposalSelector,
+        expirationManager: this.services.proposalExpirationManager,
       });
 
       allocation = await this.getAllocationFromOrder({ order, maxAgreements: 1 });
@@ -611,6 +647,7 @@ export class GolemNetwork {
         logger: this.logger,
         validateOfferProposal: order.market.offerProposalFilter,
         selectOfferProposal: order.market.offerProposalSelector,
+        expirationManager: this.services.proposalExpirationManager,
       });
 
       const maxAgreements = typeof poolSize === "number" ? poolSize : (poolSize?.max ?? poolSize?.min ?? 1);

src/market/draft-offer-proposal-pool.test.ts

@@ -7,12 +7,14 @@ describe("Draft Offer Proposal Pool", () => {
   const mockProposal = mock(OfferProposal);
   // Most of the time we're testing the case when the Proposal is in `Draft` status
   when(mockProposal.isDraft()).thenReturn(true);
+  when(mockProposal.id).thenReturn("1");
 
   // NOTE: ts-mockito instance + JS Set.add() doesn't play along, 2x instance(mockProposal) produces "the same" value for (Set.add)
 
   const secondMockProposal = mock(OfferProposal);
   // Most of the time we're testing the case when the Proposal is in `Draft` status
   when(secondMockProposal.isDraft()).thenReturn(true);
+  when(secondMockProposal.id).thenReturn("2");
 
   describe("Adding proposals", () => {
     describe("Positive cases", () => {

src/market/draft-offer-proposal-pool.ts

@@ -4,6 +4,7 @@ import { GolemMarketError, MarketErrorCode } from "./error";
 import { createAbortSignalFromTimeout, defaultLogger, Logger, runOnNextEventLoopIteration } from "../shared/utils";
 import { Observable, Subscription } from "rxjs";
 import { AcquireQueue } from "../shared/utils/acquireQueue";
+import { ExpirationManager } from "../shared/expiration/ExpirationManager";
 
 export type OfferProposalSelector = (proposals: OfferProposal[]) => OfferProposal | null;
 
@@ -28,6 +29,8 @@ export interface ProposalPoolOptions {
   minCount?: number;
 
   logger?: Logger;
+
+  expirationManager?: ExpirationManager;
 }
 
 export interface ProposalPoolEvents {
@@ -68,16 +71,19 @@ export class DraftOfferProposalPool {
   /** {@link ProposalPoolOptions.validateOfferProposal} */
   private readonly validateOfferProposal: OfferProposalFilter = (proposal: OfferProposal) => proposal !== undefined;
 
+  /** {@link ProposalPoolOptions.expirationManager} */
+  private readonly expirationManager: ExpirationManager | undefined = undefined;
+
   /**
    * The proposals that were not yet leased to anyone and are available for lease
    */
-  private available = new Set<OfferProposal>();
+  private available = new Map<OfferProposal["id"], OfferProposal>();
 
   /**
    * Returns a read-only copy of all draft offers currently in the pool
    */
   public getAvailable(): Array<OfferProposal> {
-    return [...this.available];
+    return [...this.available.values()];
   }
 
   /**
@@ -97,6 +103,13 @@ export class DraftOfferProposalPool {
       this.minCount = options.minCount;
     }
 
+    if (options?.expirationManager) {
+      this.expirationManager = options.expirationManager;
+      this.expirationManager.registerCleanupFunction((proposalId: string) => {
+        this.removeFromAvailable(proposalId);
+      });
+    }
+
     this.logger = this.logger = options?.logger || defaultLogger("proposal-pool");
   }
 
@@ -114,7 +127,7 @@ export class DraftOfferProposalPool {
       return;
     }
 
-    this.available.add(proposal);
+    this.available.set(proposal.id, proposal);
 
     this.events.emit("added", { proposal });
   }
@@ -147,7 +160,7 @@ export class DraftOfferProposalPool {
       }
       if (!this.validateOfferProposal(proposal)) {
         // Drop if not valid
-        this.removeFromAvailable(proposal);
+        this.removeFromAvailable(proposal.id);
         // and try again
         return runOnNextEventLoopIteration(tryGettingFromAvailable);
       }
@@ -158,7 +171,7 @@ export class DraftOfferProposalPool {
     // Try to get one
 
     if (proposal) {
-      this.available.delete(proposal);
+      this.available.delete(proposal.id);
       this.leased.add(proposal);
       this.events.emit("acquired", { proposal });
       return proposal;
@@ -184,7 +197,7 @@ export class DraftOfferProposalPool {
         return;
       }
       // otherwise, put it back to the list of available proposals
-      this.available.add(proposal);
+      this.available.set(proposal.id, proposal);
     } else {
       this.events.emit("removed", { proposal });
     }
@@ -196,8 +209,8 @@ export class DraftOfferProposalPool {
       this.events.emit("removed", { proposal });
     }
 
-    if (this.available.has(proposal)) {
-      this.available.delete(proposal);
+    if (this.available.has(proposal.id)) {
+      this.available.delete(proposal.id);
       this.events.emit("removed", { proposal });
     }
   }
@@ -235,8 +248,8 @@ export class DraftOfferProposalPool {
    */
   public async clear() {
     this.acquireQueue.releaseAll();
-    for (const proposal of this.available) {
-      this.available.delete(proposal);
+    for (const [id, proposal] of this.available) {
+      this.available.delete(id);
       this.events.emit("removed", { proposal });
     }
 
@@ -245,19 +258,26 @@ export class DraftOfferProposalPool {
       this.events.emit("removed", { proposal });
     }
 
-    this.available = new Set();
+    this.available = new Map();
     this.leased = new Set();
     this.events.emit("cleared");
   }
 
-  protected removeFromAvailable(proposal: OfferProposal): void {
-    this.available.delete(proposal);
-    this.events.emit("removed", { proposal });
+  protected removeFromAvailable(proposalId: OfferProposal["id"]): void {
+    const proposalToDelete = this.available.get(proposalId);
+    if (!proposalToDelete) return;
+    this.available.delete(proposalId);
+    this.events.emit("removed", { proposal: proposalToDelete });
   }
 
   public readFrom(source: Observable<OfferProposal>): Subscription {
     return source.subscribe({
-      next: (proposal) => this.add(proposal),
+      next: (proposal) => {
+        if (this.expirationManager) {
+          this.expirationManager.registerObjectForCleanup(proposal.id);
+        }
+        this.add(proposal);
+      },
       error: (err) => this.logger.error("Error while collecting proposals", err),
     });
   }

src/shared/expiration/ExpirationManager.ts

@@ -0,0 +1,87 @@
+import { Logger } from "../utils";
+
+/**
+ * Simple utility class that runs all given cleanup functions on a set interval.
+ * It can be used to manage expiration of cache entries.
+ */
+export class ExpirationManager {
+  private intervalId: NodeJS.Timeout | null = null;
+  private trackedItems: Map<string, number> = new Map();
+  private cleanupFunctions: Set<(id: string) => void | Promise<void>> = new Set();
+
+  private ttlMs: number;
+  private intervalMs: number;
+  private logger: Logger;
+
+  constructor(options: { timeToLiveMs: number; intervalMs: number; logger: Logger }) {
+    this.ttlMs = options.timeToLiveMs;
+    this.intervalMs = options.intervalMs;
+    this.logger = options.logger;
+  }
+
+  start() {
+    if (this.intervalId) return;
+    this.intervalId = setInterval(() => {
+      this.#cleanupExpiredItems().catch((error) => {
+        this.logger.error("Error occurred in ExpirationManager", { error });
+      });
+    }, this.intervalMs);
+  }
+
+  registerObjectForCleanup(id: string) {
+    if (this.trackedItems.has(id)) {
+      this.trackedItems.delete(id);
+    }
+    this.trackedItems.set(id, Date.now() + this.ttlMs);
+  }
+
+  registerCleanupFunction(fn: (id: string) => void) {
+    this.cleanupFunctions.add(fn);
+    return () => this.unregisterCleanupFunction(fn);
+  }
+
+  unregisterObjectForCleanup(id: string) {
+    this.trackedItems.delete(id);
+  }
+
+  unregisterCleanupFunction(fn: (id: string) => void) {
+    this.cleanupFunctions.delete(fn);
+  }
+
+  async #cleanupExpiredItems() {
+    const now = Date.now();
+    const expiredItems: string[] = [];
+    for (const [id, expirationDate] of this.trackedItems) {
+      if (expirationDate <= now) {
+        this.logger.debug(`Item with id ${id} has expired and will be cleaned up.`);
+        expiredItems.push(id);
+      } else {
+        // JS Map is guaranteed to maintain insertion order
+        // so when we find the first item that's not expired we
+        // can safely exit the loop
+        break;
+      }
+    }
+
+    const allProcessingPromises = expiredItems.map(async (id) => {
+      this.trackedItems.delete(id);
+      const cleanupPromises = [...this.cleanupFunctions].map((fn) => fn(id));
+      try {
+        await Promise.all(cleanupPromises);
+      } catch (error) {
+        this.logger.error(`Error occurred in cleanup functions for id ${id}`, { error, id });
+      }
+    });
+
+    await Promise.all(allProcessingPromises);
+  }
+
+  stopAndReset() {
+    if (this.intervalId) {
+      clearInterval(this.intervalId);
+      this.intervalId = null;
+    }
+    this.trackedItems.clear();
+    this.cleanupFunctions.clear();
+  }
+}

src/shared/yagna/repository/proposal-repository.ts

@@ -4,16 +4,23 @@ import { Demand, GolemMarketError, MarketErrorCode } from "../../../market";
 import { CacheService } from "../../cache/CacheService";
 import { IProposalRepository, MarketProposal } from "../../../market/proposal/market-proposal";
 import { OfferCounterProposal } from "../../../market/proposal/offer-counter-proposal";
+import { ExpirationManager } from "../../expiration/ExpirationManager";
 
 export class ProposalRepository implements IProposalRepository {
   constructor(
     private readonly marketService: MarketApi.RequestorService,
     private readonly identityService: IdentityApi.DefaultService,
     private readonly cache: CacheService<MarketProposal>,
-  ) {}
+    private readonly expirationManager: ExpirationManager,
+  ) {
+    this.expirationManager.registerCleanupFunction((proposalId: string) => {
+      this.cache.delete(proposalId);
+    });
+  }
 
   add(proposal: MarketProposal) {
     this.cache.set(proposal.id, proposal);
+    this.expirationManager.registerObjectForCleanup(proposal.id);
     return proposal;
   }