improvements to reliability from prod testing

2 files changed, 20 insertions, 7 deletions

diff --git a/python/sandcrawler/ia.py b/python/sandcrawler/ia.py
index 14716fc..3be1a79 100644
--- a/python/sandcrawler/ia.py
+++ b/python/sandcrawler/ia.py
@@ -21,6 +21,12 @@ from gwb.loader import CDXLoaderFactory
 from .misc import b32_hex, requests_retry_session, gen_file_metadata
 
 class SandcrawlerBackoffError(Exception):
+    """
+    A set of Exceptions which are raised through multiple abstraction layers to
+    indicate backpressure. For example, SPNv2 back-pressure sometimes needs to
+    be passed up through any timeout/retry code and become an actual long pause
+    or crash.
+    """
     pass
 
 ResourceResult = namedtuple("ResourceResult", [
@@ -628,7 +634,7 @@ class WaybackClient:
 class SavePageNowError(Exception):
     pass
 
-class SavePageNowBackoffError(Exception):
+class SavePageNowBackoffError(SandcrawlerBackoffError):
     pass
 
 SavePageNowResult = namedtuple('SavePageNowResult', [
@@ -702,13 +708,13 @@ class SavePageNowClient:
             },
         )
         if resp.status_code == 429:
-            raise SavePaperNowBackoffError("status_code: {}, url: {}".format(resp.status_code, request_url))
+            raise SavePageNowBackoffError("status_code: {}, url: {}".format(resp.status_code, request_url))
         elif resp.status_code != 200:
             raise SavePageNowError("SPN2 status_code: {}, url: {}".format(resp.status_code, request_url))
         resp_json = resp.json()
 
         if resp_json and 'message' in resp_json and 'You have already reached the limit of active sessions' in resp_json['message']:
-            raise SavePaperNowBackoffError(resp_json['message'])
+            raise SavePageNowBackoffError(resp_json['message'])
         elif not resp_json or 'job_id' not in resp_json:
             raise SavePageNowError(
                 "Didn't get expected 'job_id' field in SPN2 response: {}".format(resp_json))
@@ -785,13 +791,13 @@ class SavePageNowClient:
             elif status == "error:no-access":
                 status = "forbidden"
             elif status == "error:user-session-limit":
-                raise Exception("SPNv2 user-session-limit, need to backoff")
+                raise SavePageNowBackoffError("SPNv2 user-session-limit")
             elif status.startswith("error:"):
                 status = "spn2-" + status
             return ResourceResult(
                 start_url=start_url,
                 hit=False,
-                status=spn_result.status,
+                status=status,
                 terminal_url=spn_result.terminal_url,
                 terminal_dt=spn_result.terminal_dt,
                 terminal_status_code=None,
diff --git a/python/sandcrawler/workers.py b/python/sandcrawler/workers.py
index 97079ee..c290421 100644
--- a/python/sandcrawler/workers.py
+++ b/python/sandcrawler/workers.py
@@ -369,6 +369,11 @@ class KafkaJsonPusher(RecordPusher):
                 for msg in batch:
                     self.counts['total'] += 1
                     record = json.loads(msg.value().decode('utf-8'))
+                    # This complex bit of code implements backoff/backpressure
+                    # in a way that will not cause this Kafka consumer to lose
+                    # partition assignments (resulting in a rebalance). This
+                    # was needed for the ingest workers. There is probably a
+                    # better way to structure this concurrency.
                     done = False
                     while not done:
                         try:
@@ -376,12 +381,14 @@ class KafkaJsonPusher(RecordPusher):
                             break
                         except SandcrawlerBackoffError as be:
                             print("Backing off for 200 seconds: {}".format(be))
-                            self.consumer.pause()
+                            self.consumer.pause(self.consumer.assignment())
                             for i in range(40):
+                                # Beware this poll which should not be
+                                # receiving any messages because we are paused!
                                 empty_batch = self.consumer.poll(0)
                                 assert not empty_batch
                                 time.sleep(5)
-                            self.consumer.resume()
+                            self.consumer.resume(self.consumer.assignment())
                     self.counts['pushed'] += 1
                     if self.counts['total'] % 500 == 0:
                         print("Import counts: {}".format(self.worker.counts), file=sys.stderr)