Delay agreement termination until activity is destroyed

golemfactory · Apr 2, 2021 · 527d59c · 527d59c
1 parent 82fd6e3
commit 527d59c
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 8 deletions.
diff --git a/yapapi/executor/__init__.py b/yapapi/executor/__init__.py
@@ -469,8 +469,6 @@ async def start_worker(agreement: rest.market.Agreement, node_info: NodeInfo) ->
                     )
                 )
                 emit(events.WorkerFinished(agr_id=agreement.id))
-                # Don't reuse the agreement after activity creation failed
-                await agreements_pool.release_agreement(agreement.id, allow_reuse=False)
                 raise
             async with act:
                 emit(events.ActivityCreated(act_id=act.id, agr_id=agreement.id))
@@ -531,11 +529,7 @@ async def start_worker(agreement: rest.market.Agreement, node_info: NodeInfo) ->
                                         agr_id=agreement.id, exc_info=sys.exc_info()  # type: ignore
                                     )
                                 )
-                                # Don't reuse the agreement after failure
-                                await agreements_pool.release_agreement(
-                                    agreement.id, allow_reuse=False
-                                )
-                                return
+                                raise
 
             await accept_payment_for_agreement(agreement.id)
             emit(events.WorkerFinished(agr_id=agreement.id))

diff --git a/yapapi/executor/agreements_pool.py b/yapapi/executor/agreements_pool.py
@@ -57,7 +57,9 @@ async def cycle(self):
                 continue
             task = buffered_agreement.worker_task
             if task is not None and task.done():
-                await self.release_agreement(buffered_agreement.agreement.id)
+                await self.release_agreement(
+                    buffered_agreement.agreement.id, allow_reuse=task.exception() is None
+                )
 
     async def add_proposal(self, score: float, proposal: OfferProposal) -> None:
         """Adds providers' proposal to the pool of available proposals"""

diff --git a/yapapi/log.py b/yapapi/log.py
@@ -482,6 +482,13 @@ def _handle(self, event: events.Event):
                 reason = str(exc) or repr(exc) or "unexpected error"
                 self.logger.error("Error when shutting down Executor: %s", reason)
 
+        elif isinstance(event, events.AgreementTerminated):
+            if event.reason.get("golem.requestor.code") == "Success":
+                pass
+            else:
+                prov_info = self.agreement_provider_info[event.agr_id]
+                self.logger.info(f"Terminated agreement with {prov_info.name}")
+
 
 def log_summary(wrapped_emitter: Optional[Callable[[events.Event], None]] = None):
     """Output a summary of computation.