fix: violations and AI referrals are neither sent twice nor lost during an outage

includes/class-webdecoy-ai-referrals.php

@@ -135,6 +135,11 @@ public static function flush(string $apiKey): void
         if ($apiKey === '' || !function_exists('wp_remote_post')) {
             return;
         }
+        // While WebDecoy is refusing work, leave the counts where they are
+        // (WebDecoy/app#1245); they are sent under the same id afterwards.
+        if (class_exists('WebDecoy_Detection_Sender') && WebDecoy_Detection_Sender::backing_off()) {
+            return;
+        }
         global $wpdb;
         $table = self::table();
         // phpcs:ignore WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQL.InterpolatedNotPrepared
@@ -165,7 +170,9 @@ public static function flush(string $apiKey): void
 
     /**
      * True when the batch needs no retry: accepted, or refused for a reason a
-     * retry cannot fix (a 4xx, such as a key not scoped to one site).
+     * retry cannot fix (a 4xx, such as a key not scoped to one site). A 429 is
+     * "not now", not "never": the batch is kept and resent under its id, which
+     * ingest deduplicates (WebDecoy/app#1245).
      *
      * @param array<int, array{platform: string, path: string, count: int}> $referrals
      */
@@ -180,9 +187,22 @@ private static function send(string $apiKey, string $batch, array $referrals): b
             'body' => wp_json_encode(['report_id' => $batch, 'source' => 'wordpress', 'referrals' => $referrals]),
         ]);
         if (is_wp_error($response)) {
+            self::note_refusal();
             return false;
         }
         $code = (int) wp_remote_retrieve_response_code($response);
+        if ($code === 429 || $code >= 500) {
+            self::note_refusal();
+            return false;
+        }
         return $code >= 200 && $code < 500;
     }
+
+    /** Pause cloud calls after an unavailable answer (WebDecoy/app#1245). */
+    private static function note_refusal(): void
+    {
+        if (class_exists('WebDecoy_Detection_Sender')) {
+            WebDecoy_Detection_Sender::note_refusal();
+        }
+    }
 }

includes/class-webdecoy-violation-reporter.php

@@ -42,6 +42,17 @@ class WebDecoy_Violation_Reporter
     /** Hard cap on spooled rows; oldest beyond this are discarded. */
     private const MAX_QUEUE = 1000;
 
+    /**
+     * Option used as a drain lock (WebDecoy/app#1245). Every request that
+     * recorded a violation drains at shutdown, and the cron drains too; two
+     * drains at once read the same oldest rows and sent them twice.
+     * add_option() is atomic on the option name, so only one holder wins.
+     */
+    private const DRAIN_LOCK = 'webdecoy_violation_drain_lock';
+
+    /** A lock older than this belonged to a request that died; it is taken over. */
+    private const DRAIN_LOCK_TTL = 60;
+
     /** @var WebDecoy_Violation_Reporter|null */
     private static $instance = null;
 
@@ -134,7 +145,39 @@ public static function drain_queue(string $apiKey): void
         if ($apiKey === '' || !function_exists('wp_remote_post')) {
             return;
         }
+        // While WebDecoy is refusing work, keep the spool (it is capped) rather
+        // than spend its rows' attempts on requests that cannot succeed.
+        if (class_exists('WebDecoy_Detection_Sender') && WebDecoy_Detection_Sender::backing_off()) {
+            return;
+        }
+        if (!self::acquire_drain_lock()) {
+            return;
+        }
+        try {
+            self::drain_locked($apiKey);
+        } finally {
+            delete_option(self::DRAIN_LOCK);
+        }
+    }
 
+    /**
+     * Take the drain lock, or take over one left by a request that died.
+     */
+    private static function acquire_drain_lock(): bool
+    {
+        if (add_option(self::DRAIN_LOCK, (string) time(), '', 'no')) {
+            return true;
+        }
+        $held = (int) get_option(self::DRAIN_LOCK, 0);
+        if ($held > 0 && time() - $held < self::DRAIN_LOCK_TTL) {
+            return false;
+        }
+        delete_option(self::DRAIN_LOCK);
+        return add_option(self::DRAIN_LOCK, (string) time(), '', 'no');
+    }
+
+    private static function drain_locked(string $apiKey): void
+    {
         global $wpdb;
         $table = $wpdb->prefix . 'webdecoy_violation_queue';
 
@@ -171,10 +214,15 @@ public static function drain_queue(string $apiKey): void
             return;
         }
 
-        $ok = self::send($apiKey, $events);
+        $result = self::send($apiKey, $events);
 
         $idList = implode(',', array_map('intval', $ids));
-        if ($ok) {
+        if ($result === 'refused') {
+            // Ingest is unavailable, not the batch: keep its attempts for when
+            // it is back. The shared pause stops further drains meanwhile.
+            return;
+        }
+        if ($result === 'ok') {
             // phpcs:ignore WordPress.DB.DirectDatabaseQuery, WordPress.DB.PreparedSQL.InterpolatedNotPrepared
             $wpdb->query("DELETE FROM {$table} WHERE id IN ({$idList})");
         } else {
@@ -186,11 +234,13 @@ public static function drain_queue(string $apiKey): void
     }
 
     /**
-     * Blocking POST of a batch to ingest. Returns true on a 2xx response.
+     * Blocking POST of a batch to ingest: 'ok' on 2xx, 'refused' when ingest is
+     * unavailable (no answer, 429, 5xx; this also pauses cloud calls), and
+     * 'rejected' for any other answer.
      *
      * @param array<int,array<string,mixed>> $events
      */
-    private static function send(string $apiKey, array $events): bool
+    private static function send(string $apiKey, array $events): string
     {
         $response = wp_remote_post(self::ENDPOINT, [
             'timeout' => 3,
@@ -202,10 +252,13 @@ private static function send(string $apiKey, array $events): bool
             'body' => wp_json_encode(['events' => $events]),
         ]);
 
-        if (is_wp_error($response)) {
-            return false;
+        $code = is_wp_error($response) ? 0 : (int) wp_remote_retrieve_response_code($response);
+        if ($code === 0 || $code === 429 || $code >= 500) {
+            if (class_exists('WebDecoy_Detection_Sender')) {
+                WebDecoy_Detection_Sender::note_refusal();
+            }
+            return 'refused';
         }
-        $code = (int) wp_remote_retrieve_response_code($response);
-        return $code >= 200 && $code < 300;
+        return ($code >= 200 && $code < 300) ? 'ok' : 'rejected';
     }
 }

tests/HoneytokenTest.php

@@ -34,6 +34,10 @@ function get_option($k, $d = false)
 if (!function_exists('add_option')) {
     function add_option($k, $v, $a = '', $b = 'yes')
     {
+        // As in WordPress: adding an option that exists fails.
+        if (array_key_exists($k, $GLOBALS['__wd_opts'])) {
+            return false;
+        }
         $GLOBALS['__wd_opts'][$k] = $v;
         return true;
     }

tests/ViolationDrainTest.php

@@ -0,0 +1,154 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * Violation delivery when WebDecoy is unavailable or two drains overlap
+ * (WebDecoy/app#1245).
+ *
+ * Every request that records a violation drains the spool at shutdown, and the
+ * cron drains it too. Two drains at once read the same oldest rows and sent
+ * them twice; a refusal spent the rows' attempts, so an outage of two drains
+ * discarded them. Named to load after HoneytokenTest and IpEnrichmentBackoffTest,
+ * whose option and HTTP stubs this uses.
+ *
+ * Run: php tests/run.php
+ */
+
+if (!defined('ABSPATH')) {
+    define('ABSPATH', '/tmp/');
+}
+require_once dirname(__DIR__) . '/includes/class-webdecoy-detection-sender.php';
+require_once dirname(__DIR__) . '/includes/class-webdecoy-violation-reporter.php';
+
+if (!function_exists('delete_option')) {
+    function delete_option($k) { unset($GLOBALS['__wd_opts'][$k]); return true; }
+}
+if (!function_exists('wp_json_encode')) {
+    function wp_json_encode($v) { return json_encode($v); }
+}
+if (!function_exists('wp_remote_post')) {
+    function wp_remote_post($url, $args = []) {
+        $GLOBALS['wd_test_post']['calls']++;
+        return ['code' => $GLOBALS['wd_test_post']['code']];
+    }
+}
+
+/** Just enough of $wpdb for the spool queries. */
+final class WdFakeSpool
+{
+    public $prefix = 'wp_';
+    /** @var array<int,array{id:int,payload:string,attempts:int}> */
+    public $rows = [];
+
+    public function prepare($sql, ...$args) { return vsprintf(str_replace(['%d', '%s'], ['%d', "'%s'"], $sql), $args); }
+    public function get_var($sql) { return count($this->rows); }
+    public function get_results($sql)
+    {
+        $out = [];
+        foreach (array_slice($this->rows, 0, 100) as $r) {
+            $out[] = (object) ['id' => $r['id'], 'payload' => $r['payload']];
+        }
+        return $out;
+    }
+    public function delete($table, $where) { $this->rows = array_values(array_filter($this->rows, fn($r) => $r['id'] !== $where['id'])); }
+    public function query($sql)
+    {
+        if (preg_match('/DELETE FROM \S+ WHERE id IN \(([\d,]+)\)/', $sql, $m)) {
+            $ids = array_map('intval', explode(',', $m[1]));
+            $this->rows = array_values(array_filter($this->rows, fn($r) => !in_array($r['id'], $ids, true)));
+        } elseif (preg_match('/SET attempts = attempts \+ 1 WHERE id IN \(([\d,]+)\)/', $sql, $m)) {
+            $ids = array_map('intval', explode(',', $m[1]));
+            foreach ($this->rows as &$r) {
+                if (in_array($r['id'], $ids, true)) $r['attempts']++;
+            }
+        } elseif (preg_match('/WHERE attempts >= (\d+)/', $sql, $m)) {
+            $max = (int) $m[1];
+            $this->rows = array_values(array_filter($this->rows, fn($r) => $r['attempts'] < $max));
+        }
+        return true;
+    }
+}
+
+function wd_spool(int $n): WdFakeSpool
+{
+    $db = new WdFakeSpool();
+    for ($i = 1; $i <= $n; $i++) {
+        $db->rows[] = ['id' => $i, 'payload' => json_encode(['rule' => 'tripwire', 'i' => $i]), 'attempts' => 0];
+    }
+    $GLOBALS['wpdb'] = $db;
+    $GLOBALS['wd_test_post'] = ['calls' => 0, 'code' => 202];
+    unset($GLOBALS['__wd_opts']['webdecoy_violation_drain_lock']);
+    WebDecoy_Detection_Sender::reset();
+    return $db;
+}
+
+$t = ['TestRunner', 'test'];
+$same = ['TestRunner', 'assertSame'];
+$true = ['TestRunner', 'assertTrue'];
+
+$t('a delivered batch is removed from the spool', function () use ($same) {
+    $db = wd_spool(3);
+    WebDecoy_Violation_Reporter::drain_queue('key');
+    $same(1, $GLOBALS['wd_test_post']['calls']);
+    $same(0, count($db->rows));
+});
+
+$t('a drain does not run while another holds the lock', function () use ($same) {
+    $db = wd_spool(3);
+    $GLOBALS['__wd_opts']['webdecoy_violation_drain_lock'] = (string) time();
+    WebDecoy_Violation_Reporter::drain_queue('key');
+    $same(0, $GLOBALS['wd_test_post']['calls'], 'the second drain sends nothing');
+    $same(3, count($db->rows));
+});
+
+$t('a lock left by a request that died is taken over', function () use ($same) {
+    $db = wd_spool(2);
+    $GLOBALS['__wd_opts']['webdecoy_violation_drain_lock'] = (string) (time() - 120);
+    WebDecoy_Violation_Reporter::drain_queue('key');
+    $same(1, $GLOBALS['wd_test_post']['calls']);
+    $same(0, count($db->rows));
+});
+
+$t('the lock is released after a drain', function () use ($true) {
+    wd_spool(1);
+    WebDecoy_Violation_Reporter::drain_queue('key');
+    $true(!isset($GLOBALS['__wd_opts']['webdecoy_violation_drain_lock']));
+});
+
+$t('a refusal keeps the rows and their attempts, and pauses further drains', function () use ($same, $true) {
+    $db = wd_spool(3);
+    $GLOBALS['wd_test_post']['code'] = 503;
+    WebDecoy_Violation_Reporter::drain_queue('key');
+    WebDecoy_Violation_Reporter::drain_queue('key');
+    WebDecoy_Violation_Reporter::drain_queue('key');
+    $same(1, $GLOBALS['wd_test_post']['calls'], 'paused after the first refusal');
+    $same(3, count($db->rows), 'nothing is discarded during an outage');
+    $same(0, $db->rows[0]['attempts'], 'a refusal does not spend an attempt');
+    $true(WebDecoy_Detection_Sender::backing_off());
+});
+
+$t('a rejected batch still spends its attempts and is dropped at the limit', function () use ($same) {
+    $db = wd_spool(1);
+    $GLOBALS['wd_test_post']['code'] = 400;
+    WebDecoy_Violation_Reporter::drain_queue('key');
+    WebDecoy_Violation_Reporter::drain_queue('key');
+    $same(0, count($db->rows));
+    WebDecoy_Detection_Sender::reset();
+});
+
+require_once dirname(__DIR__) . '/includes/class-webdecoy-ai-referrals.php';
+
+$t('an AI referral batch refused with 429 is kept for retry and pauses cloud calls', function () use ($same, $true) {
+    $send = new ReflectionMethod('WebDecoy_AI_Referrals', 'send');
+    $send->setAccessible(true);
+    WebDecoy_Detection_Sender::reset();
+    $GLOBALS['wd_test_post'] = ['calls' => 0, 'code' => 429];
+    $same(false, $send->invoke(null, 'key', 'batch-1', [['platform' => 'chatgpt', 'path' => '/', 'count' => 1]]), '429 is not delivered');
+    $true(WebDecoy_Detection_Sender::backing_off());
+    WebDecoy_Detection_Sender::reset();
+    $GLOBALS['wd_test_post']['code'] = 400;
+    $same(true, $send->invoke(null, 'key', 'batch-1', []), 'a 4xx other than 429 cannot be fixed by retrying');
+    $GLOBALS['wd_test_post']['code'] = 202;
+    $same(true, $send->invoke(null, 'key', 'batch-1', []));
+});