From 7ed16c3c9d58d93ead1661f8a45465e3f6b19224 Mon Sep 17 00:00:00 2001
From: Tom Reitz <treitz@edanalytics.org>
Date: Mon, 12 Jun 2023 12:41:41 -0500
Subject: [PATCH] fixing a bug where the per-request timeout was including time
 the task spends in the queue waiting for an available connection from the
 pool

---
 lightbeam/api.py  |  2 +-
 lightbeam/send.py | 23 ++++++++++++-----------
 2 files changed, 13 insertions(+), 12 deletions(-)

diff --git a/lightbeam/api.py b/lightbeam/api.py
index 03aa62d..52eea66 100644
--- a/lightbeam/api.py
+++ b/lightbeam/api.py
@@ -80,7 +80,7 @@ def prepare(self, selector="*"):
     # Returns a client object with exponential retry and other parameters per configs
     def get_retry_client(self):
         return RetryClient(
-            timeout=aiohttp.ClientTimeout(total=self.lightbeam.config['connection']["timeout"]),
+            timeout=aiohttp.ClientTimeout(sock_connect=self.lightbeam.config['connection']["timeout"]),
             retry_options=ExponentialRetry(
                 attempts=self.lightbeam.config['connection']["num_retries"],
                 factor=self.lightbeam.config['connection']["backoff_factor"],
diff --git a/lightbeam/send.py b/lightbeam/send.py
index 316a984..59de9f7 100644
--- a/lightbeam/send.py
+++ b/lightbeam/send.py
@@ -87,14 +87,14 @@ async def do_send(self, endpoint):
     
     # Posts a single data payload to a single endpoint using the client
     async def do_post(self, endpoint, file_name, data, client, line, hash):
-        try:
-            status = 401
-            while status==401:
-                
-                # wait if another process has locked lightbeam while we refresh the oauth token:
-                while self.lightbeam.is_locked:
-                    await asyncio.sleep(1)
-                
+        status = 401
+        while status==401:
+            
+            # wait if another process has locked lightbeam while we refresh the oauth token:
+            while self.lightbeam.is_locked:
+                await asyncio.sleep(1)
+            
+            try:
                 async with client.post(util.url_join(self.lightbeam.api.config["data_url"], self.lightbeam.config["namespace"], endpoint),
                                         data=data,
                                         ssl=self.lightbeam.config["connection"]["verify_ssl"],
@@ -122,7 +122,8 @@ async def do_post(self, endpoint, file_name, data, client, line, hash):
                     else:
                         self.lightbeam.api.update_oauth()
         
-        except Exception as e:
-            self.lightbeam.num_errors += 1
-            self.logger.warn("{0}  (at line {1} of {2} )".format(str(e), line, file_name))
+            except Exception as e:
+                status = 400
+                self.lightbeam.num_errors += 1
+                self.logger.warn("{0}  (at line {1} of {2} )".format(str(e), line, file_name))