Ingest records immediately when next payload will exceed buffer limit (#336)

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

Author: timacdonald

agent/build/agent.phar

@@ -1 +1 @@
-0389423268858E42BBF5CE5338A2515B812A14A30FE5E79C4B19A7C3D3C45426C1061A5C34CA67300409BA190E583F1D5BFC39B7965E255675DBEA180A75A5C2
+859B5CB714BE47686CD645444022C0336586C814B1C783F48E8AD843750845E011E19E53D7CAC9C92382E1242D4A83933DF9A0686B5A9A6E0330D69100C8A047

agent/build/signature.txt

@@ -58,6 +58,16 @@ public function __construct(
 
     public function write(string $payload): void
     {
+        if ($this->buffer->isNotEmpty() && $this->buffer->willExceedThresholdWith($payload)) {
+            if ($this->sendBufferAfterDelayTimer !== null) {
+                $this->loop->cancelTimer($this->sendBufferAfterDelayTimer);
+
+                $this->sendBufferAfterDelayTimer = null;
+            }
+
+            $this->digest();
+        }
+
         $this->buffer->write($payload);
 
         if ($this->buffer->reachedThreshold()) {

agent/src/Ingest.php

@@ -14,6 +14,11 @@ public function reachedThreshold(): bool
         return false;
     }
 
+    public function willExceedThresholdWith(string $payload): bool
+    {
+        return false;
+    }
+
     /**
      * @return non-empty-string
      */

agent/src/NullBuffer.php

@@ -31,6 +31,13 @@ public function reachedThreshold(): bool
         return strlen($this->buffer) >= $this->threshold;
     }
 
+    public function willExceedThresholdWith(string $payload): bool
+    {
+        // -2 to account for the removal of the `[` and `]` characters when
+        // appending to the stream.
+        return (strlen($this->buffer) + (strlen($payload) - 2)) > $this->threshold;
+    }
+
     /**
      * @return non-empty-string
      */

agent/src/StreamBuffer.php

@@ -703,7 +703,7 @@ public function test_it_ingests_payloads_under_the_threshold_after_10_seconds():
         $ingestDetailsBrowser->assertPending([]);
     }
 
-    public function test_it_ingests_payloads_before_10_seconds_if_the_buffer_exceeds_the_threshold(): void
+    public function test_it_ingests_payloads_before_10_seconds_if_the_buffer_reaches_the_threshold(): void
     {
         $loop = new LoopFake(runForSeconds: 11);
         $server = new TcpServerFake;
@@ -724,7 +724,7 @@ public function test_it_ingests_payloads_before_10_seconds_if_the_buffer_exceeds
             ingestBrowser: $ingestBrowser,
             loop: $loop,
             server: $server,
-            maxBufferLength: 63,
+            maxBufferLength: 62,
         );
 
         $this->assertNull($e, $e?->getMessage() ?? '');
@@ -803,6 +803,68 @@ public function test_it_ingests_immediately_when_buffer_is_empty_and_a_payload_o
         $ingestDetailsBrowser->assertPending([]);
     }
 
+    public function test_it_ingests_immediately_when_incoming_payload_will_put_buffer_over_the_threshold(): void
+    {
+        $loop = new LoopFake(runForSeconds: 14);
+        $server = new TcpServerFake;
+        $ingestDetailsBrowser = new BrowserFake([
+            Response::jwt(),
+        ]);
+        $ingestBrowser = new BrowserFake([
+            Response::ingested(),
+            Response::ingested(),
+        ]);
+        $loop->addTimer(0, $server->pendingConnection([['t' => 'request']]));
+        $loop->addTimer(1, $server->pendingConnection([['t' => 'request']]));
+        $loop->addTimer(2, $server->pendingConnection([['t' => 'request']]));
+        $loop->addTimer(3, $server->pendingConnection([['t' => 'request']]));
+
+        [$output, $e] = $this->runAgent(
+            via: 'source',
+            ingestDetailsBrowser: $ingestDetailsBrowser,
+            ingestBrowser: $ingestBrowser,
+            loop: $loop,
+            server: $server,
+            maxBufferLength: 61,
+        );
+
+        $this->assertNull($e, $e?->getMessage() ?? '');
+        $this->assertLogMatches(<<<'OUTPUT'
+        {date} {info} Authentication successful {duration}
+        {date} {info} Ingest successful {duration}
+        {date} {info} Ingest successful {duration}
+        OUTPUT, $output);
+        $ingestBrowser->assertSent([
+            Request::ingest([
+                ['t' => 'request'],
+                ['t' => 'request'],
+                ['t' => 'request'],
+            ]),
+            Request::ingest([
+                ['t' => 'request'],
+            ]),
+        ]);
+        $ingestBrowser->assertProcessing([]);
+        $ingestBrowser->assertPending([]);
+        $loop->assertRun([
+            new Timer(interval: 0, runAt: 0, scheduledAt: 0, scheduledBy: $this->functionName()),
+            new Timer(interval: 1, runAt: 1, scheduledAt: 0, scheduledBy: $this->functionName()),
+            new Timer(interval: 2, runAt: 2, scheduledAt: 0, scheduledBy: $this->functionName()),
+            new Timer(interval: 3, runAt: 3, scheduledAt: 0, scheduledBy: $this->functionName()),
+            new Timer(interval: 10, runAt: 13, scheduledAt: 3, scheduledBy: 'Laravel\NightwatchAgent\Ingest::write'),
+        ]);
+        $loop->assertCanceled([
+            new Timer(interval: 10, canceledAt: 3, scheduledAt: 0, scheduledBy: 'Laravel\NightwatchAgent\Ingest::write'),
+        ]);
+        $loop->assertPending([
+            new Timer(interval: 3_600, runAt: 3_600, scheduledAt: 0, scheduledBy: 'Laravel\NightwatchAgent\IngestDetailsRepository::scheduleRefreshIn'),
+        ]);
+        $ingestDetailsBrowser->assertSent([
+            Request::json('/api/agent-auth'),
+        ]);
+        $ingestDetailsBrowser->assertPending([]);
+    }
+
     public function test_it_stops_ingesting_data_when_exceeding_quota_during_request(): void
     {
         $loop = new LoopFake(runForSeconds: 60);