Issues #12, #16, #17

README.md

@@ -14,7 +14,7 @@ further stores these incidents via [`bos-incidents`](https://github.com/PBSA/bos
 display them in the manual intervention(MINT) module [`bos-mint`](https://github.com/PBSA/bos-mint).
 
 ## Documentation
-For directions on how to install and run `bos-auto` please visit our [documentation page](http://bos-auto.readthedocs.io/en/develop/installation.html).
+For directions on how to install and run `bos-auto` please visit our [documentation page](https://www.peerplays.tech/bookie-oracle-suite-bos/).
 
 [![docs master](https://readthedocs.org/projects/bos-auto/badge/?version=latest)](http://bos-auto.rtfd.io/en/latest/)
 [![docs develop](https://readthedocs.org/projects/bos-auto/badge/?version=develop)](http://bos-auto.rtfd.io/en/develop/)

bookied/cli.py

@@ -80,12 +80,12 @@ def worker(queue):
         if q.count > 0:
             log.info("Emptying Redis Queue (default) ...")
             q.empty()
-            job = q.dequeue()
+            job = q.delete()
             while job is not None:
                 log.info("Canceling " + str(job))
                 job.cancel()
                 job.delete()
-                job = q.dequeue()
+                job = q.delete()
             log.info("Redis Queue cleared (jobs before=" + str(count_before) + ")")
         else:
             log.info("Empty Redis Queue initialized")

bookied/schedule.py

@@ -1,3 +1,4 @@
+from . import work
 import threading
 import time
 from datetime import datetime
@@ -39,6 +40,13 @@ def check_scheduled(
 
         events = []
 
+        # midway is the status of the incidents which are terminated intentionally as creating bms took time.
+        for status_name in ["midway"]:
+            for event in storage.get_events_by_call_status(
+                call=call, status_name=status_name
+            ):
+                events.append(event)
+
         for status_name in ["postponed", "unhandled exception, retrying soon"]:
             for event in storage.get_events_by_call_status(
                 call=call,
@@ -66,8 +74,10 @@ def check_scheduled(
     # Flask queue
     q = Queue(connection=get_redis())
 
-    # only push into the queue if it's somewhat empty (with 10% buffer), otherwise wait
-    if q.count + 2 < len(push_to_queue):
+    # If push_to_queue has events, then empty queue and fill it again with push_to_queue
+    # This approach ensures that events which are on the chain but not all bms are created shall be processed first.
+    if len(push_to_queue) > 0:
+        q.empty()
         for incident in push_to_queue:
             job = q.enqueue(
                 func_callback,
@@ -106,7 +116,6 @@ def run(self):
 def scheduler(delay=None, proposer=None, approver=None):  # pragma: no cover
     """
     """
-    from . import work
 
     if not delay:
         delay = config["scheduler"]["interval"]

bookied/triggers/create.py

@@ -1,10 +1,12 @@
+import time
 from . import SKIP_DYNAMIC_BMS
 from .trigger import Trigger
 from .. import exceptions
 from ..log import log
 from bookied_sync.event import LookupEvent
 from datetime import datetime
 
+
 class CreateTrigger(Trigger):
     """ This trigger inherits class:`Trigger` and deals with create incidents
         that are fired by the data proxy when new events are announced.
@@ -40,11 +42,15 @@ def _trigger(self, args):
         self.event.update()
 
         # Create Betting market Groups
-        self.createBmgs()
+        status = self.createBmgs()
 
-        return True
+        if status == 'midway':
+            return status
+        else:
+            return True
 
     def createBmgs(self):
+        tic = time.time()
         """ Go through all Betting Market groups and create them
         """
         for bmg in self.event.bettingmarketgroups:
@@ -53,32 +59,48 @@ def createBmgs(self):
             uialist = list()
             uialist = bmg["asset"]
             name = bmg["description"]["en"]
-            x = 0 
+            x = 0
             while x < len(uialist):
-                bmg["asset"] =  uialist[x]
+                bmg["asset"] = uialist[x]
                 if(len(uialist) > 1):
-                   bmg["description"]["en"] = name + "_" + str(bmg["asset"]) 
-                   # To avoid duplicate BMG's on retriggering replays from DP's
-                   if "id" in bmg:
-                      bmg["id"] = None
-                 
+                    bmg["description"]["en"] = name + "_" + str(bmg["asset"])
+                    # To avoid duplicate BMG's on retriggering replays from DP's
+                    if "id" in bmg:
+                        bmg["id"] = None
+
                 x += 1
                 # Skip dynamic bmgs
                 if bmg["dynamic"]:
-                   # Dynamic BMGs are created separately
-                   log.debug("Skipping dynamic BMG: {}".format(str(bmg.identifier)))
-                   continue
+                    # Dynamic BMGs are created separately
+                    log.debug("Skipping dynamic BMG: {}".format(str(bmg.identifier)))
+                    continue
                 bmg.update()
-                self.createBms(bmg)
+                statusCreateBms = self.createBms(bmg)
+                if statusCreateBms == "midway":
+                    print("createBmms, toc > 10")
+                    return 'midway'
+
+            # Terminate loop and return midway if bmg creation took time.
+            # So that incident is processed later.
+            toc = time.time() - tic
+            if toc > 30:
+                return 'midway'
 
     def createBms(self, bmg):
         """ Go through all betting markets and create them
         """
+        tic = time.time()
         for bm in bmg.bettingmarkets:
+
             log.debug(
                 "Updating Betting Market {} ...".format(bm["description"].get("en"))
             )
             bm.update()
+            # break the loop and return midway if bm creation takes time.
+            toc = time.time()
+            if (toc - tic) > 10:
+                return "midway"
+            tic = toc
 
     def getIncidentEvent(self):
         """ Does not throw in all cases but returns None in case of error

bookied/triggers/trigger.py

@@ -1,3 +1,4 @@
+import time
 from ..log import log
 from .. import exceptions
 from dateutil.parser import parse
@@ -104,6 +105,7 @@ def getEvent(self):
             raise exceptions.EventDoesNotExistException
 
     def trigger(self, *args, **kwargs):
+        tic = time.time()
         """ Forward a trigger to the actual trigger implementation
             in the subclass
         """
@@ -112,7 +114,8 @@ def trigger(self, *args, **kwargs):
         self.testConditions()
 
         # Execute the actual Trigger
-        self._trigger(*args, **kwargs)
+        status = self._trigger(*args, **kwargs)
+        print('trigger.py, def trigger, line 118, after _trigger done, time:', time.time() - tic)
 
         # if a proposal is going to be published, let's enable
         # blocking so we can obtain the proposal id
@@ -132,11 +135,17 @@ def trigger(self, *args, **kwargs):
         actions = [x.action() for x in transactions]
 
         # unless _trigger raises an exception
-        self.set_incident_status(
-            status_name="done", status_add=dict(proposals=proposal_ids, actions=actions)
-        )
+        if status == 'midway':
+            self.set_incident_status(
+                status_name="midway", status_add=dict(proposals=proposal_ids, actions=actions)
+            )
+            return transactions
+        else:
+            self.set_incident_status(
+                status_name="done", status_add=dict(proposals=proposal_ids, actions=actions)
+            )
 
-        return transactions
+            return transactions
 
     def normalize(self, *args, **kwargs):
         try:

bookied/web.py

@@ -180,26 +180,13 @@ def trigger():
                 ),
             )
             log.info("Forwarded incident {} to worker via redis".format(str(incident)))
-
-            # In case we "proposed" something, we also need to approve,
-            # we do that by queuing a approve
-            approve_job = q.enqueue(
-                work.approve,
-                args=(),
-                kwargs=dict(
-                    proposer=app.config.get("BOOKIE_PROPOSER"),
-                    approver=app.config.get("BOOKIE_APPROVER"),
-                ),
-            )
-
         # Return message with id
         return jsonify(
             dict(
                 result="processing",
                 message=incident,
                 id=str(job.id),
-                id_approve=str(approve_job.id),
+                id_approve=str(job.id),
             )
         )
-
     return "", 503

bookied/work.py

@@ -67,6 +67,8 @@ def process(message, **kwargs):
 
         Hence, this method has the look and feel of a dispatcher!
     """
+    approve()  # approve to minimize approval pendig map error.
+
     # We clear the buffers here so we start fresh with no
     # operation in any buffer
     lookup.clear()
@@ -262,6 +264,7 @@ def process(message, **kwargs):
     except Exception as e:
         pass
 
+    approve() # approved called after proposal
 
 #
 # Approve my own Proposals

requirements.txt

@@ -1,9 +1,9 @@
-peerplays>=0.3.7
-bos-sync>=0.3.8
-bos-incidents>=0.3.7
+peerplays
+bos-sync
+bos-incidents
 
-redis==2.10.6
-rq==0.12.0
+redis
+rq
 
 click
 click-datetime
@@ -21,7 +21,7 @@ colorlog
 tqdm
 
 # Notifications
-requests==2.21.0
+requests
 python-telegram-handler
 
 # needed for fast signing of transactions!