Bugfix/jst 648/debit note acceptance

src/payment/payments.ts

@@ -57,7 +57,7 @@ export class Payments extends EventTarget {
           this.options.invoiceFetchingInterval / 1000,
           this.lastInvoiceFetchingTime,
           this.options.maxInvoiceEvents,
-          undefined,
+          this.yagnaApi.appSessionId,
           { timeout: 0 },
         );
         for (const event of invoiceEvents) {
@@ -91,10 +91,11 @@ export class Payments extends EventTarget {
             this.options.debitNotesFetchingInterval / 1000,
             this.lastDebitNotesFetchingTime,
             this.options.maxDebitNotesEvents,
-            undefined,
+            this.yagnaApi.appSessionId,
             { timeout: 0 },
           )
           .catch(() => ({ data: [] }));
+
         for (const event of debitNotesEvents) {
           if (!this.isRunning) return;
           if (event.eventType !== "DebitNoteReceivedEvent") continue;

src/payment/rejection.ts

@@ -5,6 +5,12 @@ export enum RejectionReason {
   UnsolicitedService = "UNSOLICITED_SERVICE",
   BadService = "BAD_SERVICE",
   IncorrectAmount = "INCORRECT_AMOUNT",
+  /**
+   * We might get a debit note related to an agreement which is already covered with
+   * a final invoice. In such cases we don't want to pay for the debit note,
+   * as the payment will be already made when we accept the invoice.
+   */
+  NonPayableAgreement = "NON_PAYABLE_AGREEMENT",
 }
 
 /**

src/payment/service.ts

@@ -145,6 +145,21 @@ export class PaymentService {
   private async processDebitNote(debitNote: DebitNote) {
     try {
       if (this.paidDebitNotes.has(debitNote.id)) return;
+
+      if (!this.agreementsToPay.has(debitNote.agreementId)) {
+        const reason = {
+          rejectionReason: RejectionReason.NonPayableAgreement,
+          totalAmountAccepted: "0",
+          message:
+            "DebitNote rejected because the agreement is already covered with a final invoice that should be paid instead of the debit note",
+        };
+        await debitNote.reject(reason);
+        this.logger?.warn(
+          `DebitNote has been rejected for agreement ${debitNote.agreementId}. Reason: ${reason.message}`,
+        );
+        return;
+      }
+
       if (await this.config.debitNoteFilter(debitNote.dto)) {
         await debitNote.accept(debitNote.totalAmountDue, this.allocation!.id);
         this.paidDebitNotes.add(debitNote.id);

tests/e2e/tasks.spec.ts

@@ -223,7 +223,11 @@ describe("Task Executor", function () {
     expect(logger.logs).not.toContain("Trying to redo the task");
   });
 
-  it("should clean up the agreements in the pool if the agreement has been terminated by provider", async () => {
+  /**
+   * TODO:
+   * For the test to work properly, the midAgreementDebitNoteIntervalSec parameter (which is in the beta version) is needed, so we temporarily skip this test
+   */
+  it.skip("should clean up the agreements in the pool if the agreement has been terminated by provider", async () => {
     const eventTarget = new EventTarget();
     const executor = await TaskExecutor.create({
       package: "golem/alpine:latest",
@@ -248,10 +252,44 @@ describe("Task Executor", function () {
       // the first task should be terminated by the provider, the second one should not use the same agreement
       await executor.run(async (ctx) => console.log((await ctx.run("echo 'Hello World'")).stdout));
     } catch (error) {
-      fail(`Test failed. ${error}`);
+      throw new Error(`Test failed. ${error}`);
     } finally {
       await executor.shutdown();
     }
     expect(createdAgreementsCount).toBeGreaterThan(1);
   });
+
+  it("should only accept debit notes for agreements that were created by the executor", async () => {
+    const eventTarget1 = new EventTarget();
+    const eventTarget2 = new EventTarget();
+    const executor1 = await TaskExecutor.create("golem/alpine:latest");
+    const executor2 = await TaskExecutor.create("golem/alpine:latest");
+    const createdAgreementsIds1 = new Set();
+    const createdAgreementsIds2 = new Set();
+    const acceptedDebitNoteAgreementIds1 = new Set();
+    const acceptedDebitNoteAgreementIds2 = new Set();
+    eventTarget1.addEventListener(EventType, (event) => {
+      const ev = event as BaseEvent<unknown>;
+      if (ev instanceof Events.AgreementCreated) createdAgreementsIds1.add(ev.detail.id);
+      if (ev instanceof Events.DebitNoteAccepted) acceptedDebitNoteAgreementIds1.add(ev.detail.agreementId);
+    });
+    eventTarget2.addEventListener(EventType, (event) => {
+      const ev = event as BaseEvent<unknown>;
+      if (ev instanceof Events.AgreementCreated) createdAgreementsIds2.add(ev.detail.id);
+      if (ev instanceof Events.DebitNoteAccepted) acceptedDebitNoteAgreementIds2.add(ev.detail.agreementId);
+    });
+    try {
+      await Promise.all([
+        executor1.run(async (ctx) => console.log((await ctx.run("echo 'Executor 1'")).stdout)),
+        executor2.run(async (ctx) => console.log((await ctx.run("echo 'Executor 2'")).stdout)),
+      ]);
+    } catch (error) {
+      throw new Error(`Test failed. ${error}`);
+    } finally {
+      await executor1.shutdown();
+      await executor2.shutdown();
+    }
+    expect(acceptedDebitNoteAgreementIds1).toEqual(createdAgreementsIds1);
+    expect(acceptedDebitNoteAgreementIds2).toEqual(createdAgreementsIds2);
+  });
 });

tests/unit/payment_service.test.ts

@@ -46,14 +46,21 @@ describe("Payment Service", () => {
       await paymentService.end();
     });
 
+    /**
+     * service.end() waits for invoices to be paid, in unit-tests that should be below 5s
+     */
+    const TEST_PAYMENT_TIMEOUT_MS = 1000;
+
     it("should accept and process debit note for agreement", async () => {
       const paymentService = new PaymentService(yagnaApi, {
         logger,
+        paymentTimeout: TEST_PAYMENT_TIMEOUT_MS,
       });
       setExpectedEvents(debitNotesEvents);
       setExpectedDebitNotes(debitNotes);
       await paymentService.createAllocation();
       await paymentService.run();
+      await paymentService.acceptPayments(agreement);
       await paymentService.acceptDebitNotes(agreement.id);
       await logger.expectToInclude(`Debit Note accepted for agreement ${agreement.id}`, 100);
       await paymentService.end();
@@ -64,11 +71,13 @@ describe("Payment Service", () => {
       const paymentService = new PaymentService(yagnaApi, {
         logger,
         debitNotesFilter: alwaysRejectDebitNoteFilter,
+        paymentTimeout: TEST_PAYMENT_TIMEOUT_MS,
       });
       setExpectedEvents(debitNotesEvents);
       setExpectedDebitNotes(debitNotes);
       await paymentService.createAllocation();
       await paymentService.run();
+      await paymentService.acceptPayments(agreement);
       await paymentService.acceptDebitNotes(agreement.id);
       await logger.expectToInclude(
         `DebitNote has been rejected for agreement ${agreement.id}. Reason: DebitNote rejected by DebitNote Filter`,
@@ -77,6 +86,24 @@ describe("Payment Service", () => {
       await paymentService.end();
     });
 
+    it("should reject a debit note when the agreement is already covered with a final invoice", async () => {
+      const paymentService = new PaymentService(yagnaApi, {
+        logger,
+      });
+
+      setExpectedEvents([...invoiceEvents, ...debitNotesEvents]);
+      setExpectedDebitNotes(debitNotes);
+      setExpectedInvoices(invoices);
+      await paymentService.createAllocation();
+      await paymentService.run();
+      await paymentService.acceptDebitNotes(agreement.id);
+      await logger.expectToInclude(
+        `DebitNote has been rejected for agreement ${agreement.id}. Reason: DebitNote rejected because the agreement is already covered with a final invoice that should be paid instead of the debit note`,
+        100,
+      );
+      await paymentService.end();
+    });
+
     it("should reject when invoice rejected by Invoice Filter", async () => {
       const alwaysRejectInvoiceFilter = async () => false;
       const paymentService = new PaymentService(yagnaApi, {
@@ -100,11 +127,13 @@ describe("Payment Service", () => {
       const paymentService = new PaymentService(yagnaApi, {
         logger,
         debitNotesFilter: PaymentFilters.acceptMaxAmountDebitNoteFilter(0.00001),
+        paymentTimeout: TEST_PAYMENT_TIMEOUT_MS,
       });
       setExpectedEvents(debitNotesEvents);
       setExpectedDebitNotes(debitNotes);
       await paymentService.createAllocation();
       await paymentService.run();
+      await paymentService.acceptPayments(agreement);
       await paymentService.acceptDebitNotes(agreement.id);
       await logger.expectToInclude(
         `DebitNote has been rejected for agreement ${agreement.id}. Reason: DebitNote rejected by DebitNote Filter`,
@@ -135,11 +164,13 @@ describe("Payment Service", () => {
       const paymentService = new PaymentService(yagnaApi, {
         logger,
         debitNotesFilter: PaymentFilters.acceptMaxAmountDebitNoteFilter(7),
+        paymentTimeout: TEST_PAYMENT_TIMEOUT_MS,
       });
       setExpectedEvents(debitNotesEvents);
       setExpectedDebitNotes(debitNotes);
       await paymentService.createAllocation();
       await paymentService.run();
+      await paymentService.acceptPayments(agreement);
       await paymentService.acceptDebitNotes(agreement.id);
       await logger.expectToInclude(`Debit Note accepted for agreement ${agreement.id}`, 100);
       await paymentService.end();