Merge pull request #1167 from golemfactory/beta

Beta

Author: SewerynKras

src/activity/activity.module.ts

@@ -16,14 +16,14 @@ export interface ActivityModule {
    *
    * @return The resulting activity on the provider for further use
    */
-  createActivity(agreement: Agreement): Promise<Activity>;
+  createActivity(agreement: Agreement, signalOrTimeout?: AbortSignal | number): Promise<Activity>;
 
   /**
    * Definitely terminate any work on the provider
    *
    * @return The activity that was permanently terminated
    */
-  destroyActivity(activity: Activity): Promise<Activity>;
+  destroyActivity(activity: Activity, signalOrTimeout?: AbortSignal | number): Promise<Activity>;
 
   /**
    * Fetches the latest state of the activity. It's recommended to use this method
@@ -214,13 +214,13 @@ export class ActivityModuleImpl implements ActivityModule {
     );
   }
 
-  async createActivity(agreement: Agreement): Promise<Activity> {
+  async createActivity(agreement: Agreement, signalOrTimeout?: AbortSignal | number): Promise<Activity> {
     this.logger.debug("Creating activity", {
       agreementId: agreement.id,
       provider: agreement.provider,
     });
     try {
-      const activity = await this.activityApi.createActivity(agreement);
+      const activity = await this.activityApi.createActivity(agreement, signalOrTimeout);
       this.events.emit("activityCreated", { activity });
       this.logger.info("Created activity", {
         activityId: activity.id,
@@ -234,10 +234,10 @@ export class ActivityModuleImpl implements ActivityModule {
     }
   }
 
-  async destroyActivity(activity: Activity): Promise<Activity> {
+  async destroyActivity(activity: Activity, signalOrTimeout?: AbortSignal | number): Promise<Activity> {
     this.logger.debug("Destroying activity", activity);
     try {
-      const updated = await this.activityApi.destroyActivity(activity);
+      const updated = await this.activityApi.destroyActivity(activity, signalOrTimeout);
       this.events.emit("activityDestroyed", {
         activity: updated,
       });

src/activity/api.ts

@@ -34,9 +34,9 @@ export type ActivityEvents = {
 export interface IActivityApi {
   getActivity(id: string): Promise<Activity>;
 
-  createActivity(agreement: Agreement): Promise<Activity>;
+  createActivity(agreement: Agreement, signalOrTimeout?: AbortSignal | number): Promise<Activity>;
 
-  destroyActivity(activity: Activity): Promise<Activity>;
+  destroyActivity(activity: Activity, signalOrTimeout?: AbortSignal | number): Promise<Activity>;
 
   getActivityState(id: string): Promise<ActivityStateEnum>;
 

src/market/agreement/agreement.ts

@@ -25,7 +25,7 @@ export interface AgreementOptions {
 }
 
 export interface IAgreementRepository {
-  getById(id: string): Promise<Agreement>;
+  getById(id: string, signalOrTimeout?: AbortSignal | number): Promise<Agreement>;
 }
 
 /**

src/market/api.ts

@@ -99,7 +99,7 @@ export interface IMarketApi {
   /**
    * Retrieves an agreement based on the provided ID.
    */
-  getAgreement(id: string): Promise<Agreement>;
+  getAgreement(id: string, signalOrTimeout?: AbortSignal | number): Promise<Agreement>;
 
   /**
    * Request creating an agreement from the provided proposal
@@ -117,22 +117,30 @@ export interface IMarketApi {
    *
    * @return An agreement that's already in an "Approved" state and can be used to create activities on the Provider
    */
-  proposeAgreement(proposal: OfferProposal, options?: AgreementOptions): Promise<Agreement>;
+  proposeAgreement(
+    proposal: OfferProposal,
+    options?: AgreementOptions,
+    signalOrTimeout?: AbortSignal | number,
+  ): Promise<Agreement>;
 
   /**
    * Confirms the agreement with the provider
    */
-  confirmAgreement(agreement: Agreement, options?: AgreementOptions): Promise<Agreement>;
+  confirmAgreement(
+    agreement: Agreement,
+    options?: AgreementOptions,
+    signalOrTimeout?: AbortSignal | number,
+  ): Promise<Agreement>;
 
   /**
    * Terminates an agreement.
    */
-  terminateAgreement(agreement: Agreement, reason?: string): Promise<Agreement>;
+  terminateAgreement(agreement: Agreement, reason?: string, signalOrTimeout?: AbortSignal | number): Promise<Agreement>;
 
   /**
    * Retrieves the state of an agreement based on the provided agreement ID.
    */
-  getAgreementState(id: string): Promise<AgreementState>;
+  getAgreementState(id: string, signalOrTimeout?: AbortSignal | number): Promise<AgreementState>;
 
   /**
    * Scan the market for offers that match the given specification.

src/market/demand/demand.ts

@@ -116,4 +116,24 @@ export class Demand {
   get paymentPlatform(): string {
     return this.details.paymentPlatform;
   }
+
+  /**
+   * Demand expiration as a timestamp or null if it's not present in the properties object
+   */
+  get expiration(): number | null {
+    const expirationPropertyValue = this.details.prototype.properties.find(
+      (property) => property.key === "golem.srv.comp.expiration",
+    );
+    if (!expirationPropertyValue) {
+      return null;
+    }
+    if (typeof expirationPropertyValue.value === "number") {
+      return expirationPropertyValue.value;
+    }
+    const valuesAsNumber = Number(expirationPropertyValue.value);
+    if (Number.isNaN(valuesAsNumber)) {
+      return null;
+    }
+    return valuesAsNumber;
+  }
 }

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

@@ -5,7 +5,7 @@ import { createAbortSignalFromTimeout, defaultLogger, Logger, runOnNextEventLoop
 import { Observable, Subscription } from "rxjs";
 import { AcquireQueue } from "../shared/utils/acquireQueue";
 
-export type OfferProposalSelector = (proposals: OfferProposal[]) => OfferProposal;
+export type OfferProposalSelector = (proposals: OfferProposal[]) => OfferProposal | null;
 
 export interface ProposalPoolOptions {
   /**
@@ -48,6 +48,10 @@ export interface ProposalPoolEvents {
  * parties.
  *
  * Technically, the "market" part of you application should populate this pool with such offer proposals.
+ *
+ * It's important to know that offers are never automatically removed from the pool, even if the corresponding
+ * Demand becomes expired. It's on the application developer to ensure that a proposal is still valid before
+ * trying to sign an agreement.
  */
 export class DraftOfferProposalPool {
   public readonly events = new EventEmitter<ProposalPoolEvents>();
@@ -69,6 +73,13 @@ export class DraftOfferProposalPool {
    */
   private available = new Set<OfferProposal>();
 
+  /**
+   * Returns a read-only copy of all draft offers currently in the pool
+   */
+  public getAvailable(): Array<OfferProposal> {
+    return [...this.available];
+  }
+
   /**
    * The proposal that were already leased to someone and shouldn't be leased again
    */
@@ -121,7 +132,14 @@ export class DraftOfferProposalPool {
     const tryGettingFromAvailable = async (): Promise<OfferProposal | undefined> => {
       signal.throwIfAborted();
 
-      const proposal = this.available.size > 0 ? this.selectOfferProposal([...this.available]) : null;
+      let proposal: OfferProposal | null = null;
+      if (this.available.size > 0) {
+        try {
+          proposal = this.selectOfferProposal(this.getAvailable());
+        } catch (e) {
+          this.logger.error("Error in user-defined offer proposal selector", { error: e });
+        }
+      }
       if (!proposal) {
         // No proposal was selected, either `available` is empty or the user's proposal filter didn't select anything
         // no point retrying

src/market/market.module.ts

@@ -147,15 +147,21 @@ export interface MarketModule {
    * - ya-ts-client "wait for approval"
    *
    * @param proposal
+   * @param signalOrTimeout - The timeout in milliseconds or an AbortSignal that will be used to cancel the operation
    *
    * @return Returns when the provider accepts the agreement, rejects otherwise. The resulting agreement is ready to create activities from.
    */
-  proposeAgreement(proposal: OfferProposal): Promise<Agreement>;
+  proposeAgreement(
+    proposal: OfferProposal,
+    agreementOptions?: AgreementOptions,
+    signalOrTimeout?: AbortSignal | number,
+  ): Promise<Agreement>;
 
   /**
    * @return The Agreement that has been terminated via Yagna
+   * @param signalOrTimeout - The timeout in milliseconds or an AbortSignal that will be used to cancel the operation
    */
-  terminateAgreement(agreement: Agreement, reason?: string): Promise<Agreement>;
+  terminateAgreement(agreement: Agreement, reason?: string, signalOrTimeout?: AbortSignal | number): Promise<Agreement>;
 
   /**
    * Acquire a proposal from the pool and sign an agreement with the provider. If signing the agreement fails,
@@ -210,7 +216,7 @@ export interface MarketModule {
   /**
    * Fetch the most up-to-date agreement details from the yagna
    */
-  fetchAgreement(agreementId: string): Promise<Agreement>;
+  fetchAgreement(agreementId: string, signalOrTimeout?: AbortSignal | number): Promise<Agreement>;
 
   /**
    * Scan the market for offers that match the given demand specification.
@@ -456,8 +462,12 @@ export class MarketModuleImpl implements MarketModule {
     }
   }
 
-  async proposeAgreement(proposal: OfferProposal, options?: AgreementOptions): Promise<Agreement> {
-    const agreement = await this.marketApi.proposeAgreement(proposal, options);
+  async proposeAgreement(
+    proposal: OfferProposal,
+    options?: AgreementOptions,
+    signalOrTimeout?: AbortSignal | number,
+  ): Promise<Agreement> {
+    const agreement = await this.marketApi.proposeAgreement(proposal, options, signalOrTimeout);
 
     this.logger.info("Proposed and got approval for agreement", {
       agreementId: agreement.id,
@@ -467,8 +477,12 @@ export class MarketModuleImpl implements MarketModule {
     return agreement;
   }
 
-  async terminateAgreement(agreement: Agreement, reason?: string): Promise<Agreement> {
-    await this.marketApi.terminateAgreement(agreement, reason);
+  async terminateAgreement(
+    agreement: Agreement,
+    reason?: string,
+    signalOrTimeout?: AbortSignal | number,
+  ): Promise<Agreement> {
+    await this.marketApi.terminateAgreement(agreement, reason, signalOrTimeout);
 
     this.logger.info("Terminated agreement", {
       agreementId: agreement.id,
@@ -665,8 +679,8 @@ export class MarketModuleImpl implements MarketModule {
     }
   }
 
-  async fetchAgreement(agreementId: string): Promise<Agreement> {
-    return this.marketApi.getAgreement(agreementId);
+  async fetchAgreement(agreementId: string, signalOrTimeout?: AbortSignal | number): Promise<Agreement> {
+    return this.marketApi.getAgreement(agreementId, signalOrTimeout);
   }
 
   /**

src/network/api.ts

@@ -20,33 +20,43 @@ export interface INetworkApi {
   /**
    * Creates a new network with the specified options.
    * @param options NetworkOptions
+   * @param signalOrTimeout - The timeout in milliseconds or an AbortSignal that will be used to cancel the operation
    */
-  createNetwork(options: NetworkOptions): Promise<Network>;
+  createNetwork(options: NetworkOptions, signalOrTimeout?: AbortSignal | number): Promise<Network>;
 
   /**
    * Removes an existing network.
    * @param network - The network to be removed.
+   * @param signalOrTimeout - The timeout in milliseconds or an AbortSignal that will be used to cancel the operation
    */
-  removeNetwork(network: Network): Promise<void>;
+  removeNetwork(network: Network, signalOrTimeout?: AbortSignal | number): Promise<void>;
 
   /**
    * Creates a new node within a specified network.
    * @param network - The network to which the node will be added.
    * @param nodeId - The ID of the node to be created.
    * @param nodeIp - Optional IP address for the node. If not provided, the first available IP address will be assigned.
+   * @param signalOrTimeout - The timeout in milliseconds or an AbortSignal that will be used to cancel the operation
    */
 
-  createNetworkNode(network: Network, nodeId: string, nodeIp?: string): Promise<NetworkNode>;
+  createNetworkNode(
+    network: Network,
+    nodeId: string,
+    nodeIp?: string,
+    signalOrTimeout?: AbortSignal | number,
+  ): Promise<NetworkNode>;
 
   /**
    * Removes an existing node from a specified network.
    * @param network - The network from which the node will be removed.
    * @param node - The node to be removed.
+   * @param signalOrTimeout - The timeout in milliseconds or an AbortSignal that will be used to cancel the operation
    */
-  removeNetworkNode(network: Network, node: NetworkNode): Promise<void>;
+  removeNetworkNode(network: Network, node: NetworkNode, signalOrTimeout?: AbortSignal | number): Promise<void>;
 
   /**
    * Returns the identifier of the requesor
+   * @param signalOrTimeout - The timeout in milliseconds or an AbortSignal that will be used to cancel the operation
    */
-  getIdentity(): Promise<string>;
+  getIdentity(signalOrTimeout?: AbortSignal | number): Promise<string>;
 }

src/network/network.module.test.ts

@@ -24,7 +24,7 @@ describe("Network", () => {
       networkApi: instance(mockNetworkApi),
       logger: instance(imock<Logger>()),
     });
-    when(mockNetworkApi.createNetwork(anything())).thenResolve(instance(mockNetwork));
+    when(mockNetworkApi.createNetwork(anything(), anything())).thenResolve(instance(mockNetwork));
     when(mockNetwork.getNetworkInfo()).thenReturn({
       id: "test-id-1",
       ip: "192.168.0.0",
@@ -188,13 +188,13 @@ describe("Network", () => {
     it("should remove network", async () => {
       const network = instance(mockNetwork);
       await networkModule.removeNetwork(network);
-      verify(mockNetworkApi.removeNetwork(network)).once();
+      verify(mockNetworkApi.removeNetwork(network, anything())).once();
     });
 
     it("should not remove network that doesn't exist", async () => {
       const network = instance(mockNetwork);
       const mockError = new Error("404");
-      when(mockNetworkApi.removeNetwork(network)).thenReject(mockError);
+      when(mockNetworkApi.removeNetwork(network, anything())).thenReject(mockError);
       await expect(networkModule.removeNetwork(network)).rejects.toMatchError(mockError);
     });
   });