diff --git a/CHANGELOG.md b/CHANGELOG.md index fde7b01..df8866d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ --- develop --- +* fix: Persist filter enabled state on fresh and existing installations, show enabled/disabled filter and schedule status using theme colours, and exclude disabled filters from report and schedule selection * fix: Make bundled Net_DNS2 constructor nullability explicit for PHP 8.4 without changing accepted arguments * issue: Shorten the graph drilldown button title from 'View NetFlow Traffic In Range' to 'NetFlow' so it reads cleanly as a label in Cacti's graph action menu diff --git a/README.md b/README.md index 8de558c..1ddac5b 100644 --- a/README.md +++ b/README.md @@ -211,3 +211,37 @@ resolution. ----------------------------------------------- Copyright (c) 2004-2026 - The Cacti Group, Inc. + +### Filter and schedule status + +Filters and schedules show Enabled in green and Disabled in grey, using the Cacti theme status classes. A disabled filter is excluded from report and schedule selectors and does not run scheduled reports. Disabling a filter does not stop its listener or delete history. Installations missing the filter state column retain existing filters as enabled when the column is created. + +Missing filter-state columns are repaired by the background upgrade process, including same-version repairs. Pages return a retry message until repair completes; scheduled reports defer without issuing schema changes. Editing a schedule retains its existing disabled filter and labels it Disabled. + +Pending filter-state migration leaves scheduled queue entries and lastsent/start timestamps unchanged. FlowView uses its own dispatcher on both Cacti versions: deferred or failed children retain their original occurrence, without saving a replacement. Native queue IDs and the older report-ID convention are both accepted. If resetting the retained row to pending fails, the row is kept and a queue-repair error is logged. + +Queue dispatch processes a bounded snapshot, attempting each occurrence once per pass. A failed occurrence stays retained while healthy entries continue; previously attempted rows sort behind new work so repeated failures cannot monopolize the next snapshot. Schema loss or exhausted concurrency stops the pass. + +Scheduler capacity checks remove exact child registrations only when the operating system confirms that their PID no longer exists. Live or uncertain processes (including permission errors and platforms without POSIX liveness probes) retain their capacity reservation. Failed legacy queue resets now log an explicit queue-repair error while retaining the original occurrence. + + +FlowView uses one dispatcher on both supported Cacti versions. A child records +`complete` after successful preparation/delivery and saved schedule timestamps, +or `retry` only for a known pre-delivery deferral. These markers let a later +poller reconcile a child that finishes after its parent's timeout. Completed +occurrences are deleted without executing them again. Only pending work or an +explicit retry marker can start a new child. + +Every started child requires a recorded outcome, including an exit code of zero. +A timeout, crash or early exit with no recorded outcome leaves the occurrence `running` and +logs its queue ID instead of automatically resending it. If it remains unresolved, +check the child process and report/notification logs, then reconcile the specific +`reports_queued` row: remove an already-delivered occurrence only after repairing +its schedule timestamps, or return a confirmed undelivered occurrence to pending. +Do not reset an unresolved row merely because its process has disappeared. + +Completion cleanup and executable reports use separate bounded snapshots, so +persistent cleanup failures cannot consume the pending-work budget. Cleanup +still runs first and does not require a free execution slot. + +Scheduled FlowView notifications use one observable delivery helper on Cacti 1.2 and 1.3. A failed queue lookup before delivery requests retry. A returned mailer error, invalid notification configuration, or failed report-log write holds the occurrence for manual reconciliation without advancing schedule timestamps. This includes partial delivery to multiple notification targets; it is never automatically resent. diff --git a/flowview.php b/flowview.php index 0a2a903..b86a345 100644 --- a/flowview.php +++ b/flowview.php @@ -34,6 +34,7 @@ require_once($config['base_path'] . '/lib/time.php'); flowview_connect(); +flowview_require_filter_status(); set_default_action(); @@ -321,7 +322,7 @@ function load_session_for_filter() { WHERE id = ?', array(get_request_var('query'))); - if (cacti_sizeof($query)) { + if (cacti_sizeof($query) && ($query['enabled'] ?? '') === 'on') { foreach($query as $column => $value) { switch($column) { case 'name': diff --git a/flowview_filters.php b/flowview_filters.php index ca758a5..8559397 100644 --- a/flowview_filters.php +++ b/flowview_filters.php @@ -34,6 +34,7 @@ require_once($config['base_path'] . '/lib/timespan_settings.php'); flowview_connect(); +flowview_require_filter_status(); set_default_action(); @@ -382,6 +383,10 @@ function clearFilter() { 'align' => 'right', 'sort' => 'ASC' ), + 'nosort2' => array( + 'display' => __('Status', 'flowview'), + 'align' => 'center' + ), 'nosort1' => array( 'display' => __('Resolution', 'flowview'), 'align' => 'right', @@ -421,6 +426,8 @@ function clearFilter() { form_selectable_cell($type, $filter['id']); form_selectable_cell($filter['id'], $filter['id'], '', 'right'); form_selectable_cell($sort, $filter['id'], '', 'right'); + $status = flowview_filter_status($filter); + form_selectable_cell('' . $status['label'] . '', $filter['id'], '', 'center'); form_selectable_cell($filter['resolve'], $filter['id'], '', 'right'); form_checkbox_cell($filter['name'], $filter['id']); diff --git a/flowview_schedules.php b/flowview_schedules.php index 77b3ea7..3501b0e 100644 --- a/flowview_schedules.php +++ b/flowview_schedules.php @@ -32,6 +32,7 @@ require_once($config['base_path'] . '/plugins/flowview/includes/functions.php'); flowview_connect(); +flowview_require_filter_status(); set_default_action(); @@ -56,12 +57,7 @@ $formats = reports_get_format_files(); -$query_array = array_rekey( - flowview_db_fetch_assoc('SELECT id, name - FROM plugin_flowview_queries - ORDER BY name'), - 'id', 'name' -); +$query_array = flowview_schedule_filters((int) get_filter_request_var('id')); if (db_table_exists('plugin_notification_lists')) { $notification_lists = array_rekey( @@ -1038,7 +1034,8 @@ function clearFilter() { form_selectable_cell('-', $row['id'], '', 'right'); } - form_selectable_cell(($row['enabled'] == 'on' ? "" . __('Yes', 'flowview') . "":"" . __('No', 'flowview') . ""), $row['id'], '', 'right'); + $status = flowview_filter_status($row); + form_selectable_cell('' . $status['label'] . '', $row['id'], '', 'right'); form_checkbox_cell($row['name'], $row['id']); form_end_row(); diff --git a/flowview_upgrade.php b/flowview_upgrade.php index a0e8224..f02bc91 100644 --- a/flowview_upgrade.php +++ b/flowview_upgrade.php @@ -23,11 +23,12 @@ +-------------------------------------------------------------------------+ */ -chdir('../../'); +chdir(__DIR__ . '/../../'); require('./include/cli_check.php'); require_once('./plugins/flowview/includes/functions.php'); require_once('./plugins/flowview/setup.php'); require_once('./plugins/flowview/includes/database.php'); +require_once('./plugins/flowview/includes/filter_upgrade.php'); flowview_connect(); @@ -81,25 +82,36 @@ $info = plugin_flowview_version(); $current_version = $info['version']; -/* do a version check */ -if ($forcever == '' && $old_version == $current_version) { - cacti_log('Your Flowview is already up to date (v' . $current_version . ' vs v' . $old_version . ') not upgrading. Use --forcever to override', true, 'FLOWVIEW'); - exit(0); -} - -if (!register_process_start('flowview', 'upgrade', 1, 86400)) { +$registered = register_process_start('flowview', 'upgrade', 1, 86400); +if (!$registered) { if (empty($forcever)) { print 'WARNING: Running process detected. To override use --forcever' . PHP_EOL; exit(1); } } +if ($registered) { + $upgrade_pid = getmypid(); + register_shutdown_function(static function () use ($upgrade_pid) { + unregister_process('flowview', 'upgrade', 1, $upgrade_pid); + }); +} + +if (!flowview_filter_status_upgrade()) { + cacti_log('FlowView filter state upgrade failed; it will be retried.', true, 'FLOWVIEW'); + exit(1); +} + +/* do a version check */ +if ($forcever == '' && $old_version == $current_version) { + cacti_log('Your Flowview is already up to date (v' . $current_version . ' vs v' . $old_version . ') not upgrading. Use --forcever to override', true, 'FLOWVIEW'); + exit(0); +} + cacti_log('Upgrading from v' . $old_version . ' to ' . $current_version, true, 'FLOWVIEW'); flowview_upgrade($current_version, $old_version); -unregister_process('flowview', 'upgrade', 1); - exit(0); /** diff --git a/includes/database.php b/includes/database.php index b457a30..be112cc 100644 --- a/includes/database.php +++ b/includes/database.php @@ -514,3 +514,8 @@ function flowview_db_get_table_column_types($table, mixed $cnn_id = false) { return $cols; } + +/** Check live metadata without Cacti's per-request column cache. */ +function flowview_filter_status_ready(): bool { + return (bool) flowview_db_fetch_row("SHOW COLUMNS FROM plugin_flowview_queries LIKE 'enabled'", false); +} diff --git a/includes/filter_upgrade.php b/includes/filter_upgrade.php new file mode 100644 index 0000000..73ed2a6 --- /dev/null +++ b/includes/filter_upgrade.php @@ -0,0 +1,18 @@ + $data) { - switch($type) { - case 'email': - if (!isset($data['to_email'])) { - cacti_log(sprintf("WARNING: Email Report '%s' not sent! Missing 'to_email' attribute in request", $report['name']), false, 'REPORTS'); - break; - } else { - $to_email = $data['to_email']; - } - - if (isset($data['cc_email'])) { - $cc_email = $data['cc_email']; - } else { - $cc_email = ''; - } - - if (isset($data['bcc_email'])) { - $bcc_email = $data['bcc_email']; - } else { - $bcc_email = ''; - } - - if (isset($data['reply_to'])) { - $reply_to = $data['reply_to']; - } else { - $reply_to = ''; - } - - mailer($from, $to_email, $cc_email, $bcc_email, $reply_to, $subject, $oput_html, $oput_text, $attachments, $headers); - - break; - case 'notification_list': - if (!isset($data['id'])) { - cacti_log(sprintf("WARNING: Email Report '%s' not sent! Missing notification list 'id' attribute in request", $report['name']), false, 'REPORTS'); - break; - } else { - $list = db_fetch_row_prepared('SELECT * - FROM plugin_notify_list - WHERE id = ?', - [$data['id']]); - - if (cacti_sizeof($list)) { - /* process the format file */ - $report_tag = ''; - $theme = 'modern'; - $output_html = ''; - $format = $list['format_file'] != '' ? $list['format_file']:'default'; - - $format_ok = reports_load_format_file($format, $output, $report_tag, $theme); - - flowview_debug('Format File Loaded, Format is ' . ($format_ok ? 'Ok':'Not Ok') . ', Report Tag is ' . $report_tag); - - if ($format_ok) { - if ($report_tag) { - $oput_html = str_replace('', $oput_raw, $output); - } else { - $oput_html = $output . PHP_EOL . $oput_raw; - } - } else { - $oput_html = $oput_raw; - } - - $to_email = $list['emails']; - $cc_emails = isset($list['cc_emails']) ? $list['cc_emails']:''; - $bcc_emails = $list['bcc_emails']; - $reply_to = isset($list['reply_to']) ? $list['reply_to']:''; - - mailer($from, $to_email, $cc_emails, $bcc_emails, $reply_to, $subject, $oput_html, $oput_text, $attachments, $headers); - } else { - cacti_log(sprintf("WARNING: Email Report '%s' not sent! Unable to locate notification list '%s'", $report['name'], $id), false, 'REPORTS'); - } - } - - break; - default: - cacti_log(sprintf("WARNING: Email Report '%s' not sent! Unknown notification type '%s' attribute in request", $report['name'], $type), false, 'REPORTS'); - break; - } - } - } - - $end_time = microtime(true); - - $save = []; - - $save['id'] = 0; - $save['name'] = $report['name']; - $save['source'] = $source; - $save['source_id'] = $source_id; - $save['report_output_type'] = $report_type; - $save['report_raw_data'] = json_encode($raw_data); - $save['report_raw_output'] = $oput_raw; - $save['report_html_output'] = $oput_html; - $save['report_txt_output'] = $oput_text; - $save['send_type'] = $report['request_type']; - $save['send_time'] = date('Y-m-d H:i:s'); - $save['run_time'] = $end_time - $start_time; - $save['sent_by'] = $report['requested_by']; - $save['sent_id'] = $report['requested_id']; - $save['notification'] = $report['notification']; - - sql_save($save, 'reports_log'); - } +function reports_log_and_notify($id, $start_time, $report_type, $source, $source_id, $subject, &$raw_data, &$oput_raw, &$oput_html, &$oput_text, $attachments = [], $headers = []): ?bool { + return flowview_reports_log_and_notify($id, $start_time, $report_type, $source, $source_id, $subject, $raw_data, $oput_raw, $oput_html, $oput_text, $attachments, $headers); } /** @@ -262,14 +115,16 @@ function reports_queue($name, $request_type, $source, $source_id, $command, $not function reports_run($id): bool { global $config; - require_once($config['base_path'] . '/lib/poller.php'); - $report = db_fetch_row_prepared('SELECT * FROM reports_queued WHERE id = ?', [$id]); if (cacti_sizeof($report)) { + if ($report['source'] === 'flowview') { + return flowview_reports_run((int) $id, $report); + } + require_once($config['base_path'] . '/lib/poller.php'); db_execute_prepared('UPDATE reports_queued SET status = ?, start_time = ? WHERE id = ?', @@ -280,14 +135,14 @@ function reports_run($id): bool { $start = microtime(true); - $return_code = 0; + $return_code = 1; $output = []; $command = $report['run_command'] . " --report-id=$id"; $timeout = $report['run_timeout']; cacti_log("The report:$id has command was:$command"); - $last_line = exec_with_timeout($command, $output, $return_code, $timeout); + exec_with_timeout($command, $output, $return_code, $timeout); $end = microtime(true); diff --git a/includes/functions.php b/includes/functions.php index bbe9f3f..5f427c3 100644 --- a/includes/functions.php +++ b/includes/functions.php @@ -23,6 +23,8 @@ +-------------------------------------------------------------------------+ */ +require_once(__DIR__ . '/report_delivery.php'); + // reports_run()/reports_queue()/reports_log_and_notify() became native in // Cacti 1.3. Whether core will declare them is fixed by the Cacti version at // bootstrap, so gate the pre-1.3 shims on that rather than on function_exists(): @@ -730,6 +732,7 @@ function flowview_display_filter(): void { = $start_part && $partition <= $end_part) { - $inc_tables[$t['table']]['table_name'] = $t['table']; - if (!isset($_SESSION['sess_flowview_table_details'][$t['table']])) { $details = flowview_db_fetch_row("SELECT MIN(start_time) AS min_date, MAX(end_time) AS max_date, '' AS table_partition FROM {$t['table']}"); - if (!cacti_sizeof($details)) { - cacti_log("WARNING: No Details for {$t['table']} returned"); - } - - $_SESSION['sess_flowview_table_details'][$t['table']] = $details; } else { $details = $_SESSION['sess_flowview_table_details'][$t['table']]; } + if (!$details) { + unset($_SESSION['sess_flowview_table_details'][$t['table']]); + cacti_log("WARNING: No Details for {$t['table']} returned"); + return false; + } + // MIN/MAX return NULL for an empty partition. Recheck it on the + // next request rather than caching empty bounds while it grows. + if ($details['min_date'] === null && $details['max_date'] === null) { + unset($_SESSION['sess_flowview_table_details'][$t['table']]); + continue; + } + $_SESSION['sess_flowview_table_details'][$t['table']] = $details; + $inc_tables[$t['table']]['table_name'] = $t['table']; + $inc_tables[$t['table']]['table_partition'] = $details['table_partition']; $inc_tables[$t['table']]['min_date'] = $details['min_date']; $inc_tables[$t['table']]['max_date'] = $details['max_date']; @@ -2753,6 +2787,10 @@ function run_flow_query($session, $query_id, $start, $end) { WHERE id = ?', [$query_id]); + if (!$data) { + return false; + } + $title = flowview_db_fetch_cell_prepared('SELECT name FROM plugin_flowview_queries WHERE id = ?', @@ -3410,6 +3448,9 @@ function run_flow_query($session, $query_id, $start, $end) { } $tables = get_tables_for_query($start, $end); + if ($tables === false) { + return false; + } $sql = ''; $all_params = []; $results = []; @@ -3561,6 +3602,13 @@ function run_flow_query($session, $query_id, $start, $end) { } } + if (!is_array($results)) { + return false; + } + if (!$results) { + return ['data' => [], 'table' => '', 'title' => $title]; + } + $output = $data; $output['data'] = $results; @@ -4220,10 +4268,10 @@ function parallel_database_query_run($requests) { raise_message_javascript('parallel_query_timeout', __('The Parallel Query Timed Out. Please contact your administrator', 'flowview'), MESSAGE_LEVEL_ERROR); foreach($requests as $request_id) { - parallel_database_query_cleanup($request_id); + parallel_database_query_cleanup($request_id, true); } - exit; + return false; } if (cacti_sizeof($requests) == 1) { @@ -4237,16 +4285,29 @@ function parallel_database_query_run($requests) { WHERE id = ?', [$requests[0]]); - $tables = array_rekey( - flowview_db_fetch_assoc('SELECT map_table, map_create, results - FROM parallel_database_query - WHERE id IN(' . implode(', ', $requests) . ')'), - 'map_table', ['map_create', 'results'] - ); + $query = json_decode($reduce_query ?: '', true); + $tables = flowview_db_fetch_assoc('SELECT map_table, map_create, results + FROM parallel_database_query + WHERE id IN(' . implode(', ', $requests) . ')'); + + $valid_results = is_array($tables) && count($tables) === count($requests); + if ($valid_results) { + foreach ($tables as $index => $data) { + $tables[$index]['results'] = json_decode($data['results'] ?: '', true); + if (!is_array($tables[$index]['results'])) { + $valid_results = false; + break; + } + } + } - if ($reduce_query != '' && cacti_sizeof($tables)) { + // Every component must succeed before any results are materialized or reduced. + if (is_array($query) && isset($query['sql_query'], $query['sql_where'], $query['sql_groupby'], $query['sql_having'], $query['sql_order'], $query['sql_limit'], $query['sql_params']) + && is_array($query['sql_params']) && $valid_results) { + $tables = array_rekey($tables, 'map_table', ['map_create', 'results']); $i = 0; $temp_table = ''; + $materialized = true; foreach($tables as $table => $data) { if ($i == 0) { @@ -4255,10 +4316,13 @@ function parallel_database_query_run($requests) { $create = str_replace($table, $temp_table, $data['map_create']); $create = str_replace('CREATE TABLE', 'CREATE TEMPORARY TABLE', $create); - flowview_db_execute($create); + if (flowview_db_execute($create) === false) { + $materialized = false; + break; + } } - $results = json_decode($data['results'], true); + $results = $data['results']; if (cacti_sizeof($results)) { $sql = []; @@ -4288,14 +4352,15 @@ function parallel_database_query_run($requests) { //cacti_log($sql_prefix . implode(', ', $sql)); /* insert entries into the intermediary table */ - flowview_db_execute($sql_prefix . implode(', ', $sql)); + if (flowview_db_execute($sql_prefix . implode(', ', $sql)) === false) { + $materialized = false; + break; + } } $i++; } - $query = json_decode($reduce_query, true); - $final_query = "{$query['sql_query']} FROM $temp_table {$query['sql_where']} @@ -4306,18 +4371,19 @@ function parallel_database_query_run($requests) { //cacti_log($final_query); - $results = flowview_db_fetch_assoc_prepared($final_query, $query['sql_params']); + $results = $materialized ? flowview_db_fetch_assoc_prepared($final_query, $query['sql_params']) : false; //cacti_log(json_encode($results)); flowview_db_execute("DROP TEMPORARY TABLE IF EXISTS $temp_table"); } else { - $results = []; + $results = false; } } foreach($requests as $request_id) { - parallel_database_query_cleanup($request_id); + // Failed cached requests must be recreated, with new shards, on retry. + parallel_database_query_cleanup($request_id, !is_array($results)); } return $results; @@ -7337,3 +7403,178 @@ function get_set_default_fast_engine() { return $myengine; } + +/** + * Return the theme status class and translated state without missing-key warnings. + * + * @param array $record Filter or schedule row. + * @return array + */ +function flowview_filter_status(array $record): array { + if (!array_key_exists('enabled', $record)) { + return ['class' => 'deviceDisabled', 'label' => __('Unavailable', 'flowview')]; + } + + return $record['enabled'] === 'on' + ? ['class' => 'deviceUp', 'label' => __('Enabled', 'flowview')] + : ['class' => 'deviceDisabled', 'label' => __('Disabled', 'flowview')]; +} + +/** Enabled choices plus the existing association when editing a schedule. */ +function flowview_schedule_filters(int $schedule_id): array { + $rows = flowview_db_fetch_assoc_prepared('SELECT id, name, enabled FROM plugin_flowview_queries + WHERE enabled = "on" OR id = (SELECT query_id FROM plugin_flowview_schedules WHERE id = ?) + ORDER BY name', [$schedule_id]); + $options = []; + foreach ($rows ?: [] as $row) { + $options[$row['id']] = $row['enabled'] === 'on' ? $row['name'] : $row['name'] . ' (' . __('Disabled', 'flowview') . ')'; + } + return $options; +} + +/** Resolve native queue IDs and the pre-1.3 report-ID convention. */ +function flowview_schedule_queue_id(array $options) { + $value = $options['queue-id'] ?? false; + if ($value === false) { + $value = $options['report-id'] ?? false; + } + return filter_var($value, FILTER_VALIDATE_INT, ['options' => ['min_range' => 1]]) ?: false; +} + +/** Own FlowView dispatch so failed/deferred children cannot consume queue rows. */ +function flowview_reports_run(int $id, ?array $report = null): bool { + global $config; + if (!flowview_filter_status_ready()) { + return false; + } + $report ??= db_fetch_row_prepared('SELECT * FROM reports_queued WHERE id=? AND source=?', [$id, 'flowview']); + if (!$report) { + return false; + } + if (($report['status'] ?? '') === 'complete') { + return flowview_reports_complete($id, (int) $report['source_id']); + } + if (($report['status'] ?? '') === 'retry') { + if (!db_execute_prepared('UPDATE reports_queued SET status=? WHERE id=? AND source=? AND status=?', ['pending', $id, 'flowview', 'retry'])) { + return false; + } + } + require_once $config['base_path'] . '/lib/poller.php'; + if (!db_execute_prepared('UPDATE reports_queued SET status=?, start_time=? WHERE id=? AND source=? AND status=?', ['running', date('Y-m-d H:i:s'), $id, 'flowview', 'pending']) || db_affected_rows() !== 1) { + return false; + } + $output = []; + $status = -2; // Sentinel: unchanged only when no process could be started. + $command = $report['run_command'] . ' --report-id=' . (int) $report['source_id'] . ' --queue-id=' . $id; + $timeout = (int) $report['run_timeout']; + if ($timeout <= 0) { + $timeout = max(1, (int) read_config_option('scheduler_timeout') ?: 300); + } + exec_with_timeout($command, $output, $status, $timeout); + $expected = 'running'; + if ($status !== -2) { + // Exit status alone cannot prove completion, even when it is zero. + $outcome = db_fetch_row_prepared('SELECT status FROM reports_queued WHERE id=? AND source=?', [$id, 'flowview']); + if (($outcome['status'] ?? '') === 'complete') { + return flowview_reports_complete($id, (int) $report['source_id']); + } + if (($outcome['status'] ?? '') !== 'retry') { + cacti_log('FlowView occurrence ' . $id . ' has an unresolved child outcome (exit ' . $status . '); retained without retry to prevent duplicate delivery. Check the child and reconcile reports_queued before retrying.', false, 'FLOWVIEW'); + return false; + } + $expected = 'retry'; + } + // Never overwrite a retry another dispatcher has already claimed. + if (!db_execute_prepared('UPDATE reports_queued SET status=? WHERE id=? AND source=? AND status=?', ['pending', $id, 'flowview', $expected])) { + cacti_log('Unable to reset the retained FlowView occurrence; queue repair is required.', false, 'FLOWVIEW'); + } + return false; +} + +/** Delete a delivered occurrence without executing its child again. */ +function flowview_reports_complete(int $id, int $schedule_id): bool { + // Delivery succeeded: retry only cleanup, never preparation or notification. + for ($attempt = 0; $attempt < 2; $attempt++) { + if (db_execute_prepared('DELETE FROM reports_queued WHERE id=? AND source=?', [$id, 'flowview']) !== false) { + return true; + } + } + cacti_log('Unable to remove completed FlowView occurrence ' . $id . ' for schedule ' . $schedule_id . '; queue repair is required. The report already completed; remove this reports_queued row without rerunning it.', false, 'FLOWVIEW'); + return false; +} + +/** Attempt a bounded, fair queue snapshot without retrying failures in the same pass. */ +function flowview_reports_run_pending(int $concurrency = 1, int $limit = 100): int { + if (!flowview_filter_status_ready()) { + return 0; + } + $limit = max(1, min(1000, $limit)); + $done = 0; + // Cleanup has its own budget: persistent deletion failures cannot starve execution. + $completed = db_fetch_assoc_prepared('SELECT id,source_id FROM reports_queued WHERE source=? AND status=? ORDER BY start_time,id LIMIT ' . $limit, ['flowview', 'complete']); + if (is_array($completed)) { + foreach ($completed as $report) { + if (flowview_reports_complete((int) $report['id'], (int) $report['source_id'])) { + $done++; + } + } + } else { + cacti_log('Unable to read completed FlowView occurrences for cleanup.', false, 'FLOWVIEW'); + } + // Executable work has an independent bounded, fair snapshot. + $reports = db_fetch_assoc_prepared('SELECT id FROM reports_queued WHERE source=? AND status IN (?, ?) ORDER BY start_time,scheduled_time,id LIMIT ' . $limit, ['flowview', 'pending', 'retry']); + if (!is_array($reports)) { + cacti_log('Unable to read pending FlowView occurrences.', false, 'FLOWVIEW'); + return $done; + } + foreach ($reports as $report) { + if (!flowview_filter_status_ready()) { + break; + } + $running = flowview_running_schedule_children(); + if ($running === null || (int) $running >= max(1, $concurrency)) { + break; + } + if (flowview_reports_run((int) $report['id'])) { + $done++; + } + } + return $done; +} + +/** Count live/uncertain children, removing only exact registrations confirmed dead. */ +function flowview_running_schedule_children(): ?int { + $rows = db_fetch_assoc_prepared('SELECT taskname,taskid,pid FROM processes WHERE tasktype=? AND taskname LIKE ?', ['flowsched', 'child%']); + if (!is_array($rows)) { + return null; + } + $running = 0; + foreach ($rows as $row) { + if (flowview_schedule_pid_alive($row['pid']) !== false) { + $running++; + continue; + } + // A concurrent registration with another PID must survive this cleanup. + $deleted = db_execute_prepared('DELETE FROM processes WHERE tasktype=? AND taskname=? AND taskid=? AND pid=?', ['flowsched', $row['taskname'], $row['taskid'], $row['pid']]); + if (!$deleted) { + $running++; + cacti_log('Unable to remove a dead FlowView child registration.', false, 'FLOWVIEW'); + } elseif ((int) db_affected_rows() !== 1) { + // A replacement PID may now own the reservation; do not grant its slot. + $running++; + } + } + return $running; +} + +/** Null means liveness cannot be established; permission errors never mean dead. */ +function flowview_schedule_pid_alive($pid): ?bool { + $valid = filter_var($pid, FILTER_VALIDATE_INT, ['options' => ['min_range' => 2, 'max_range' => 2147483647]]); + if (!$valid || !function_exists('posix_kill') || !function_exists('posix_get_last_error')) { + return null; + } + if (@posix_kill($valid, 0)) { + return true; + } + return posix_get_last_error() === 3 ? false : null; +} diff --git a/includes/report_delivery.php b/includes/report_delivery.php new file mode 100644 index 0000000..743f811 --- /dev/null +++ b/includes/report_delivery.php @@ -0,0 +1,80 @@ + read_config_option('settings_from_email') ?: 'cacti@cacti.net', + 'name' => read_config_option('settings_from_name') ?: 'Cacti ' . ucfirst($source) + ]; + $oput_text ??= ''; + foreach ($notifications as $type => $data) { + if (!is_array($data)) { + return flowview_delivery_unresolved($id); + } + $html = $oput_html; + if ($type === 'email' && (!empty($data['to_email']) || !empty($data['cc_email']) || !empty($data['bcc_email']))) { + $to = $data['to_email'] ?? ''; + $cc = $data['cc_email'] ?? ''; + $bcc = $data['bcc_email'] ?? ''; + $reply = $data['reply_to'] ?? ''; + } elseif ($type === 'notification_list' && !empty($data['id'])) { + $list = db_fetch_row_prepared('SELECT * FROM plugin_notification_lists WHERE id = ?', [$data['id']]); + if (!$list || (empty($list['emails']) && empty($list['cc_emails']) && empty($list['bcc_emails']))) { + return flowview_delivery_unresolved($id); + } + $output = ''; + $tag = false; + $theme = 'modern'; + if (reports_load_format_file(($list['format_file'] ?? '') ?: 'default', $output, $tag, $theme)) { + $html = $tag ? str_replace('', $oput_raw, $output) : $output . PHP_EOL . $oput_raw; + } else { + $html = $oput_raw; + } + $to = $list['emails'] ?? ''; + $cc = $list['cc_emails'] ?? ''; + $bcc = $list['bcc_emails'] ?? ''; + $reply = $list['reply_to'] ?? ($data['reply_to'] ?? ''); + } else { + return flowview_delivery_unresolved($id); + } + // Cacti's mailer returns an empty error string on success. An error + // can follow partial delivery, so it must never produce a safe retry. + $error = mailer($data['from'] ?? $from, $to, $cc, $bcc, $reply, $subject, $html, $oput_text, $attachments, $headers); + if ($error !== '') { + return flowview_delivery_unresolved($id); + } + } + $save = [ + 'id' => 0, 'name' => $report['name'], 'source' => $source, 'source_id' => $source_id, + 'report_output_type' => $report_type, 'report_raw_data' => json_encode($raw_data), + 'report_raw_output' => $oput_raw, 'report_html_output' => $oput_html, 'report_txt_output' => $oput_text, + 'send_type' => $report['request_type'], 'send_time' => date('Y-m-d H:i:s'), + 'run_time' => microtime(true) - $start_time, 'sent_by' => $report['requested_by'], + 'sent_id' => $report['requested_id'], 'notification' => $report['notification'] + ]; + if (!sql_save($save, 'reports_log')) { + return flowview_delivery_unresolved($id); + } + return true; +} + +function flowview_delivery_unresolved($id): ?bool { + cacti_log('FlowView occurrence ' . $id . ' notification or report-log failure; delivery may be partial. Hold for repair and verify delivery before resending.', false, 'FLOWVIEW'); + return null; +} diff --git a/locales/po/cacti.pot b/locales/po/cacti.pot index 23efd5d..796c69f 100644 --- a/locales/po/cacti.pot +++ b/locales/po/cacti.pot @@ -2794,3 +2794,11 @@ msgstr "" #: setup.php msgid "NetFlow" msgstr "" + +#: includes/functions.php +msgid "Unavailable" +msgstr "" + +#: includes/functions.php +msgid "FlowView filter schema upgrade is pending. Please try again shortly." +msgstr "" diff --git a/phpunit.xml b/phpunit.xml index 2ec8407..6139bae 100644 --- a/phpunit.xml +++ b/phpunit.xml @@ -38,7 +38,10 @@ Net/DNS2/Resolver.php Net/DNS2/Updater.php includes/functions.php + includes/functions-pre13.php + includes/report_delivery.php includes/database.php + includes/filter_upgrade.php diff --git a/run_schedule.php b/run_schedule.php index 17c052c..ce8407e 100644 --- a/run_schedule.php +++ b/run_schedule.php @@ -43,11 +43,13 @@ $force = false; $scheduled = false; $report_id = false; +$queue_id = false; $shortopts = 'VvHh'; $longopts = [ 'scheduled::', 'report-id::', + 'queue-id::', 'debug', 'force', 'version', @@ -73,6 +75,9 @@ case 'report-id': $report_id = is_array($value) ? end($value) : $value; + break; + case 'queue-id': + $queue_id = is_array($value) ? end($value) : $value; break; case 'version': display_version(); @@ -86,12 +91,22 @@ } } +// Do not claim pending rows while the filter-state schema is unavailable. +if ($scheduled && !flowview_filter_status_ready()) { + cacti_log('FlowView scheduled reports deferred until the filter-state upgrade completes.', false, 'FLOWVIEW'); + exit(75); +} + if ($scheduled == true) { /* silently end if the registered process is still running, or process table missing */ if (!$force) { if (!register_process_start('flowsched', 'master', 0, 1200)) { exit(0); } + $master_pid = getmypid(); + register_shutdown_function(static function () use ($master_pid) { + unregister_process('flowsched', 'master', 0, $master_pid); + }); } $start = microtime(true); @@ -103,66 +118,59 @@ $concurrent_processes = 1; } - $reports = db_fetch_assoc_prepared('SELECT * - FROM reports_queued - WHERE status = ? - AND source = ?', - ['pending', 'flowview']); - - while(true) { - $running = db_fetch_cell_prepared('SELECT COUNT(*) - FROM processes - WHERE taskname = ? - AND tasktype LIKE ?', - ['flowsched', 'child%']); - - if ($running < $concurrent_processes) { - $report = db_fetch_row_prepared('SELECT * - FROM reports_queued - WHERE source = ? - AND status = ? - LIMIT 1', - ['flowview', 'pending']); - - if (cacti_sizeof($report)) { - reports_run($report['id']); - - $run++; - } else { - break; - } - - sleep(1); - } - } + $run = flowview_reports_run_pending((int) $concurrent_processes); $end = microtime(true); cacti_log(sprintf('FLOWVIEW REPORT STATS: Time:%0.2f Reports:%s', $end - $start, $run), false, 'SYSTEM'); - if (!$force) { - unregister_process('flowsched', 'master', 0); - } } else { + $report_id = flowview_schedule_queue_id(['report-id' => $report_id, 'queue-id' => $queue_id]); if ($report_id === false) { cacti_log('FATAL: Flowview Schedule Report ID not provided', true, 'FLOWVIEW'); exit(1); } + $id = null; + $outcome = null; + $child_pid = getmypid(); + $owns_registration = false; + register_shutdown_function(static function () use (&$outcome, &$owns_registration, $child_pid, &$id, $report_id) { + if ($outcome !== null && !db_execute_prepared('UPDATE reports_queued SET status=? WHERE id=? AND source=? AND status=?', [$outcome, $report_id, 'flowview', 'running'])) { + cacti_log('Unable to record FlowView child outcome for occurrence ' . $report_id . '; queue repair is required. Do not resend without verifying delivery.', false, 'FLOWVIEW'); + } + if ($owns_registration) { + unregister_process('flowsched', "child_$id", 0, $child_pid); + } + }); + $report = db_fetch_row_prepared('SELECT rq.* FROM reports_queued AS rq WHERE rq.source = ? AND rq.id = ?', ['flowview', $report_id]); + if ($report === false) { + $outcome = 'retry'; + exit(75); + } + if (cacti_sizeof($report)) { $id = $report['source_id']; + if (!flowview_filter_status_ready()) { + // The FlowView parent retains this original occurrence on a retry exit. + $outcome = 'retry'; + exit(75); + } + /* silently end if the registered process is still running, or process table missing */ if (!$force) { if (!register_process_start('flowsched', "child_$id", 0, 1200)) { - exit(0); + $outcome = 'retry'; + exit(75); } + $owns_registration = true; } $t = time(); @@ -174,13 +182,31 @@ WHERE id = ?', [$id]); + if ($schedule === false) { + $outcome = 'retry'; + exit(75); + } + if (cacti_sizeof($schedule)) { - flowview_db_execute_prepared('UPDATE plugin_flowview_schedules + $delivery = plugin_flowview_run_schedule($id, $report_id); + if ($delivery === null) { + cacti_log('FlowView occurrence ' . $report_id . ' has an unresolved notification or log failure; verify delivery before repairing the queue.', false, 'FLOWVIEW'); + exit(76); + } + if ($delivery === false) { + // A deferred or failed preparation must preserve the occurrence + // and schedule timestamps just like the initial readiness check. + $outcome = 'retry'; + exit(75); + } + + if (!flowview_db_execute_prepared('UPDATE plugin_flowview_schedules SET lastsent = ?, start = ? WHERE id = ?', - array($r, date('Y-m-d H:i:s'), $id)); - - plugin_flowview_run_schedule($id, $report_id); + array($r, date('Y-m-d H:i:s'), $id))) { + cacti_log('FlowView occurrence ' . $report_id . ' finished delivery but schedule timestamps could not be saved; queue repair is required. Verify delivery and repair the schedule before clearing the occurrence.', false, 'FLOWVIEW'); + exit(76); + } $end = microtime(true); @@ -189,9 +215,7 @@ cacti_log('FLOWVIEW SCHEDULE STATS: ' . $cacti_stats , true, 'SYSTEM'); } - if (!$force) { - unregister_process('flowsched', "child_$id", 0); - } + $outcome = 'complete'; } else { cacti_log("WARNING: Unable to find Report ID $report_id", false, 'REPORTS'); } diff --git a/setup.php b/setup.php index 8726c0f..39d7acb 100644 --- a/setup.php +++ b/setup.php @@ -130,7 +130,7 @@ function plugin_flowview_upgrade(): bool { function plugin_flowview_check_upgrade(bool $force = false): void { global $config; - $files = ['plugins.php', 'flowview.php', 'index.php']; + $files = ['plugins.php', 'flowview.php', 'flowview_filters.php', 'flowview_schedules.php', 'index.php']; if (isset($_SERVER['PHP_SELF']) && !in_array(basename($_SERVER['PHP_SELF']), $files)) { return; } @@ -146,7 +146,7 @@ function plugin_flowview_check_upgrade(bool $force = false): void { FROM plugin_config WHERE directory = ?', array('flowview')); - if ($current != $old || $force) { + if ($current != $old || $force || !flowview_filter_status_ready()) { $php_binary = read_config_option('path_php_binary'); exec_background($php_binary, '-q ' .$config['base_path'] . '/plugins/flowview/flowview_upgrade.php'); @@ -158,6 +158,17 @@ function plugin_flowview_check_upgrade(bool $force = false): void { } } +/** Keep ordinary page requests out of metadata-lock waits during repair. */ +function flowview_require_filter_status(): void { + if (flowview_filter_status_ready()) { + return; + } + plugin_flowview_check_upgrade(); + http_response_code(503); + print html_escape(__('FlowView filter schema upgrade is pending. Please try again shortly.', 'flowview')); + exit; +} + /** * @return mixed */ @@ -1147,6 +1158,7 @@ function flowview_setup_table(): void { flowview_db_execute("CREATE TABLE IF NOT EXISTS `plugin_flowview_queries` ( id int(11) unsigned NOT NULL AUTO_INCREMENT, name varchar(255) NOT NULL, + enabled char(2) NOT NULL DEFAULT 'on', device_id int(11) unsigned NOT NULL, template_id int(11) NOT NULL DEFAULT '0', ex_addr varchar(46) NOT NULL DEFAULT '', diff --git a/tests/README.md b/tests/README.md index e938b89..fa15d2c 100644 --- a/tests/README.md +++ b/tests/README.md @@ -38,3 +38,26 @@ connection assertions. Security tests validate PHP 8.2 syntax compatibility and the plugin's install/version/uninstall structure without executing plugin source. +The filter lifecycle regressions also launch the actual web and upgrade entrypoints +in subprocesses with isolated core API fixtures. They verify that all three web +pages do not load the schema-mutation helper and defer repair, including +same-version installations, and that the +background worker registers before attempting repair. The queue regression uses +an in-memory SQLite database to check persisted ordering across multiple passes: +a full failed batch remains pending while healthy rows advance on the next pass. +This queue test requires PDO SQLite and skips explicitly when it is unavailable; +it does not validate MariaDB schema syntax or metadata locking. + + +Timeout regressions load Cacti's actual process helper and persist queue state in +SQLite while a worker writes stderr and completes after the parent timeout. Both +legacy and native entrypoints must avoid requeueing and reconcile late completion +without a second delivery. Separate actual-child tests cover durable outcome +markers, PID-scoped shutdown cleanup, deferred preparation and timestamp-write +failure after completion. Fixtures adapt database APIs and worker delivery; they +do not test SMTP or replace live MariaDB concurrency validation. + +Successful-dispatch doubles explicitly supply recorded child completion; an exit +code alone never synthesizes that outcome. Regressions exercise an actual child +connection failure that exits zero before its outcome callback, and a persistent +queue with 100 undeletable completed rows followed by executable work. diff --git a/tests/Unit/FilterLifecycleRegressionTest.php b/tests/Unit/FilterLifecycleRegressionTest.php new file mode 100644 index 0000000..dcc5208 --- /dev/null +++ b/tests/Unit/FilterLifecycleRegressionTest.php @@ -0,0 +1,364 @@ +fixtureRoot = sys_get_temp_dir() . '/flowview-lifecycle-' . bin2hex(random_bytes(8)); + mkdir($this->fixtureRoot . '/plugins/flowview', 0700, true); + mkdir($this->fixtureRoot . '/include'); + mkdir($this->fixtureRoot . '/lib'); + $plugin = dirname(__DIR__, 2); + foreach (['setup.php', 'includes', 'flowview.php', 'flowview_filters.php', 'flowview_schedules.php'] as $file) { + symlink($plugin . '/' . $file, $this->fixtureRoot . '/plugins/flowview/' . $file); + } + foreach (['flowview_upgrade.php', 'run_schedule.php'] as $file) { + copy($plugin . '/' . $file, $this->fixtureRoot . '/plugins/flowview/' . $file); + } + file_put_contents($this->fixtureRoot . '/plugins/flowview/INFO', "[info]\nversion=fixture-version\n"); + foreach (['auth.php', 'cli_check.php'] as $file) { + copy(dirname(__DIR__) . '/fixtures/filter-lifecycle.php', $this->fixtureRoot . '/include/' . $file); + } + foreach (['poller.php', 'time.php', 'timespan_settings.php'] as $file) { + file_put_contents($this->fixtureRoot . '/lib/' . $file, 'runFixture = function ($command, $mode = 'pending', $nativeCore = false, $recipients = null) { + $env = getenv(); + if ($nativeCore) { + global $config; + $source = getenv('FLOWVIEW_NATIVE_REPORTS_SOURCE') ?: $config['base_path'] . '/lib/reports.php'; + $text = file_get_contents($source); + $start = strpos($text, 'function reports_log_and_notify('); + if ($start === false) { $this->markTestSkipped('Native Cacti notification helper not available'); } + // Load the actual official helper in isolation, without core's unrelated declarations. + $end = strpos($text, "\nfunction reports_queue(", $start); + expect($end)->not->toBeFalse(); + $env['FLOWVIEW_FIXTURE_NATIVE_HELPER'] = $this->fixtureRoot . '/native-notify.php'; + file_put_contents($env['FLOWVIEW_FIXTURE_NATIVE_HELPER'], 'fixtureRoot . '/result.json'; + $process = proc_open($command, [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes, $this->fixtureRoot . '/plugins/flowview', $env); + fclose($pipes[0]); + $output = stream_get_contents($pipes[1]); + $error = stream_get_contents($pipes[2]); + fclose($pipes[1]); + fclose($pipes[2]); + $exit = proc_close($process); + expect($error)->toBe(''); + return [$exit, $output]; + }; +}); +afterEach(function () { + $iterator = new RecursiveIteratorIterator(new RecursiveDirectoryIterator($this->fixtureRoot, FilesystemIterator::SKIP_DOTS), RecursiveIteratorIterator::CHILD_FIRST); + foreach ($iterator as $file) { + if ($file->isDir() && !$file->isLink()) { + rmdir($file->getPathname()); + } else { + unlink($file->getPathname()); + } + } + rmdir($this->fixtureRoot); +}); +it('serves a retry response without DDL from every guarded web entrypoint', function ($page, $ready) { + [$exit, $output] = ($this->runFixture)([PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/' . $page], $ready ? 'ready' : 'pending'); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + $actions = array_column($result['calls'], 0); + expect($exit)->toBe(0)->and($actions)->not->toContain('ddl') + ->and($result['repair_loaded'])->toBeFalse(); + if ($ready) { + expect($output)->toBe('')->and($actions)->toContain('page-ready')->not->toContain('background'); + } else { + expect($result['status'])->toBe(503)->and($output)->toContain('schema upgrade is pending') + ->and($actions)->not->toContain('page-ready'); + $background = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'background')); + expect($background)->toHaveCount(1)->and($background[0][1][1])->toContain('/plugins/flowview/flowview_upgrade.php'); + } +})->with(['flowview.php', 'flowview_filters.php', 'flowview_schedules.php'])->with([false, true]); +it('repairs same-version schemas only after background registration and reports repair failure', function ($mode, $code, $writes, $force = false) { + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/flowview_upgrade.php']; + if ($force) { + $command[] = '--forcever=fixture-version'; + } + [$exit, $output] = ($this->runFixture)($command, $mode); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + expect($result)->toBeArray($output); + $actions = array_column($result['calls'], 0); + expect($exit)->toBe($code)->and($result['repair_loaded'])->toBeTrue() + ->and($actions[0])->toBe('register') + ->and(array_filter($actions, fn ($action) => $action === 'ddl'))->toHaveCount($writes); + $released = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'unregister')); + expect($released)->toBe($mode === 'busy' ? [] : [['unregister', ['flowview', 'upgrade', 1, $result['pid']]]]); +})->with([['pending', 0, 1], ['failed', 1, 1], ['ready', 0, 0], ['busy', 1, 0], ['pending', 0, 1, true], ['busy', 0, 1, true]]); +it('retains a full failing snapshot while advancing to healthy rows on the next pass', function () { + if (!extension_loaded('pdo_sqlite')) { + $this->markTestSkipped('PDO SQLite is needed for the persisted queue regression'); + } + [$exit, $output] = ($this->runFixture)([PHP_BINARY, dirname(__DIR__) . '/fixtures/filter-queue.php', dirname(__DIR__, 2), $this->fixtureRoot]); + $result = json_decode($output, true); + expect($exit)->toBe(0)->and($result['first'])->toBe(0)->and($result['second'])->toBe(2)->and($result['third'])->toBe(0) + ->and($result['attempts'])->toBe([1, 2, 3, 4, 1, 2, 6]) + ->and($result['retryRace'])->toBeFalse()->and($result['raceStatus'])->toBe(['status' => 'running']) + ->and($result['lostClaim'])->toBeFalse()->and($result['claimedStatus'])->toBe(['status' => 'running']) + ->and($result['afterFirst'])->toHaveCount(4) + ->and($result['remaining'])->toBe([['id' => 1, 'status' => 'pending'], ['id' => 2, 'status' => 'pending']]); +}); + +it('preserves late schema failures through the real child and parent completion path', function ($native, $retry) { + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/run_schedule.php', '--report-id=' . ($native ? 7 : 51)]; + if ($native) { + $command[] = '--queue-id=51'; + } + [$exit, $output] = ($this->runFixture)($command, $retry ? 'late' : 'child-disabled'); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + $writes = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'write')); + $released = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'unregister')); + expect($output)->toBe('')->and($exit)->toBe($retry ? 75 : 0) + ->and($released)->toBe([['unregister', ['flowsched', 'child_7', 0, $result['pid']]]]) + ->and($writes)->toHaveCount($retry ? 1 : 2); + expect(end($writes)[2])->toBe([$retry ? 'retry' : 'complete', 51, 'flowview', 'running']); + if (!$retry) { + expect($writes[0][1])->toContain('UPDATE plugin_flowview_schedules'); + } + global $config; + $oldBase = $config['base_path']; + $config['base_path'] = $this->fixtureRoot; + try { + plugin_test_reset(); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 60]); + plugin_test_queue_db_result('exec_with_timeout', $exit); + plugin_test_queue_db_result('report_outcome', ['status' => $exit === 0 ? 'complete' : 'retry']); + expect(flowview_reports_run(51))->toBe(!$retry); + $deletes = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => str_starts_with($c['sql'], 'DELETE FROM reports_queued'))); + expect($deletes)->toHaveCount($retry ? 0 : 1); + } finally { + $config['base_path'] = $oldBase; + } +})->with([false, true])->with([false, true]); + +it('returns retry from the real child when a queue, schedule or filter lookup fails', function ($mode, $native) { + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/run_schedule.php', '--report-id=' . ($native ? 7 : 51)]; + if ($native) { + $command[] = '--queue-id=51'; + } + [$exit] = ($this->runFixture)($command, $mode); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + expect($exit)->toBe(75); + $actions = array_column($result['calls'], 0); + $writes = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'write')); + expect($writes)->toHaveCount(1); + if ($writes) { + expect($writes[0][2])->toBe(['retry', 51, 'flowview', 'running']); + } + if ($mode !== 'report-read-failure') { + $released = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'unregister')); + expect($released)->toBe([['unregister', ['flowsched', 'child_7', 0, $result['pid']]]]); + } +})->with(['report-read-failure', 'schedule-read-failure', 'filter-read-failure'])->with([false, true]); + +it('retains a timed out parallel report through the actual child and parent', function ($native) { + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/run_schedule.php', '--report-id=' . ($native ? 7 : 51)]; + if ($native) { + $command[] = '--queue-id=51'; + } + [$exit, $output] = ($this->runFixture)($command, 'parallel-timeout'); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + expect($exit)->toBe(75)->and($output)->toBe(''); + $writes = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'write')); + expect(array_filter($writes, fn ($c) => str_contains($c[1], 'UPDATE plugin_flowview_schedules')))->toBe([]); + $cleanup = array_values(array_filter($writes, fn ($c) => str_starts_with($c[1], 'DELETE FROM parallel_database_query_shard'))); + expect(array_column($cleanup, 2))->toBe([[71]]); + $released = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'unregister')); + expect($released)->toBe([['unregister', ['flowsched', 'child_7', 0, $result['pid']]]]); + global $config; + $oldBase = $config['base_path']; + $config['base_path'] = $this->fixtureRoot; + try { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 60]); + plugin_test_queue_db_result('exec_with_timeout', $exit); + plugin_test_queue_db_result('report_outcome', ['status' => $exit === 0 ? 'complete' : 'retry']); + expect(flowview_reports_run(51))->toBeFalse(); + $deletes = array_filter($GLOBALS['__test_db_calls'], fn ($c) => str_starts_with($c['sql'], 'DELETE FROM reports_queued')); + expect($deletes)->toBe([]); + } finally { + $config['base_path'] = $oldBase; + } +})->with([false, true]); + +it('recreates shards and launches the same query after a parallel timeout', function () { + if (!extension_loaded('pdo_sqlite')) { + $this->markTestSkipped('PDO SQLite is needed for the persisted parallel retry regression'); + } + [$exit, $output] = ($this->runFixture)([PHP_BINARY, dirname(__DIR__) . '/fixtures/parallel-retry.php', dirname(__DIR__, 2), $this->fixtureRoot]); + $result = json_decode($output, true); + expect($exit)->toBe(0)->and($result['timeout'])->toBeFalse() + ->and($result['remaining'])->toBe(0)->and($result['remainingShards'])->toBe(0) + ->and($result['retry'])->not->toBe($result['first'])->and($result['retryShards'])->toBe(1) + ->and($result['result'])->toBe([])->and($result['launches'])->toBe([$result['first'], $result['retry']]) + ->and($result['cached'])->toBe($result['retry'])->and($result['cachedResult'])->toBe([]); +}); + +it('completes an exit-zero report with stderr using Cactis actual process helper', function ($runner) { + global $config; + file_put_contents($this->fixtureRoot . '/warning.php', 'runFixture)([PHP_BINARY, dirname(__DIR__) . '/fixtures/report-stderr.php', $config['base_path'], dirname(__DIR__, 2), $this->fixtureRoot, $runner]); + $result = json_decode($output, true); + expect($exit)->toBe(0)->and($result['helperResult'])->toBeFalse()->and($result['status'])->toBe(0) + ->and($result['success'])->toBeTrue()->and($result['writes'])->toHaveCount(2) + ->and($result['writes'][1]['sql'])->toStartWith('DELETE FROM reports_queued'); +})->with(['native', 'legacy']); + +it('reconciles a timeout with stderr and late completion without resending on either dispatcher', function ($runner) { + if (!extension_loaded('pdo_sqlite')) { + $this->markTestSkipped('PDO SQLite is needed for the persisted timeout regression'); + } + global $config; + [$exit, $output] = ($this->runFixture)([PHP_BINARY, dirname(__DIR__) . '/fixtures/report-timeout.php', $config['base_path'], dirname(__DIR__, 2), $this->fixtureRoot, $runner]); + $result = json_decode($output, true); + expect($exit)->toBe(0)->and($result['remaining'])->toBe([])->and($result['deliveries'])->toHaveCount(1); + $resets = array_filter($result['writes'], fn ($c) => ($c[1][0] ?? '') === 'pending'); + expect($resets)->toBe([]); + $claims = array_filter($result['writes'], fn ($c) => str_contains($c[0], 'start_time=?')); + expect($claims)->toHaveCount(1); +})->with(['native', 'legacy']); + +it('holds a delivered occurrence when schedule timestamp persistence fails', function ($native) { + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/run_schedule.php', '--report-id=' . ($native ? 7 : 51)]; + if ($native) { $command[] = '--queue-id=51'; } + [$exit] = ($this->runFixture)($command, 'timestamp-failure'); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + expect($exit)->toBe(76); + $outcomes = array_filter($result['calls'], fn ($c) => $c[0] === 'write' && str_contains($c[1], 'UPDATE reports_queued')); + expect($outcomes)->toBe([]); + $released = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'unregister')); + expect($released)->toBe([['unregister', ['flowsched', 'child_7', 0, $result['pid']]]]); +})->with([false, true]); + +it('releases only an acquired scheduler master registration at shutdown', function ($mode, $force) { + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/run_schedule.php', '--scheduled']; + if ($force) { $command[] = '--force'; } + [$exit] = ($this->runFixture)($command, $mode); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + $released = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'unregister')); + expect($exit)->toBe(0)->and($released)->toBe($mode === 'master-busy' || $force ? [] : [['unregister', ['flowsched', 'master', 0, $result['pid']]]]); +})->with([['ready', false], ['master-busy', false], ['ready', true]]); + +it('retains a child connection failure even when the actual entrypoint exits zero', function ($native) { + file_put_contents($this->fixtureRoot . '/connection.php', 'fixtureRoot . '/plugins/flowview/run_schedule.php', '--report-id=' . ($native ? 7 : 51)]; + if ($native) { $command[] = '--queue-id=51'; } + [$exit] = ($this->runFixture)($command, 'connection-failure'); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + expect($exit)->toBe(0)->and(array_column($result['calls'], 0))->toContain('connect')->not->toContain('write', 'register'); + global $config; + $oldBase = $config['base_path']; + $config['base_path'] = $this->fixtureRoot; + try { + plugin_test_reset(); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 1]); + plugin_test_queue_db_result('exec_with_timeout', $exit); + expect($native ? flowview_reports_run(51) : reports_run(51))->toBeFalse(); + $writes = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'db_execute_prepared')); + expect($writes)->toHaveCount(1)->and($writes[0]['params'][0])->toBe('running'); + } finally { $config['base_path'] = $oldBase; } +})->with([false, true]); + +it('runs pending work beyond a full snapshot of undeletable completed occurrences', function () { + if (!extension_loaded('pdo_sqlite')) { $this->markTestSkipped('PDO SQLite is needed for the bounded cleanup regression'); } + [$exit, $output] = ($this->runFixture)([PHP_BINARY, dirname(__DIR__) . '/fixtures/completion-budget.php', dirname(__DIR__, 2), $this->fixtureRoot]); + $result = json_decode($output, true); + expect($exit)->toBe(0)->and($result['done'])->toBe(1)->and($result['attempts'])->toBe([101]) + ->and($result['deletes'])->toHaveCount(201)->and($result['remaining'])->toHaveCount(100) + ->and(array_unique(array_column($result['remaining'], 'status')))->toBe(['complete']); +}); + + +it('retries only a pre-delivery notification lookup failure in the actual child', function ($native, $mode) { + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/run_schedule.php', '--report-id=' . ($native ? 7 : 51)]; + if ($native) { $command[] = '--queue-id=51'; } + // The post-delivery fixture deliberately throws; suppress its expected fatal diagnostic. + if ($mode === 'notification-after-delivery') { + array_splice($command, 1, 0, ['-d', 'display_errors=0', '-d', 'log_errors=0']); + } + [$exit] = ($this->runFixture)($command, $mode); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + $writes = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'write')); + $actions = array_column($result['calls'], 0); + $queueReads = array_filter($result['calls'], fn ($c) => $c[0] === 'row' && str_contains($c[1], 'reports_queued')); + expect($queueReads)->toHaveCount(2); + if ($mode === 'notification-after-delivery') { + expect($exit)->toBe(255, json_encode($result))->and($actions)->toContain('delivery')->and($writes)->toBe([]); + } else { + expect($exit)->toBe(75)->and($actions)->not->toContain('delivery', 'save-log') + ->and($writes)->toHaveCount(1)->and($writes[0][2])->toBe(['retry', 51, 'flowview', 'running']); + } + $released = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'unregister')); + expect($released)->toBe([['unregister', ['flowsched', 'child_7', 0, $result['pid']]]]); +})->with([false, true])->with(['notification-read-failure', 'notification-missing', 'notification-after-delivery']); + + +it('holds returned notification errors instead of retrying or completing', function ($native, $mode, $nativeCore) { + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/run_schedule.php', '--report-id=' . ($native ? 7 : 51)]; + if ($native) { $command[] = '--queue-id=51'; } + [$exit] = ($this->runFixture)($command, $mode, $nativeCore); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + $writes = array_filter($result['calls'], fn ($c) => $c[0] === 'write'); + expect($exit)->toBe(76)->and($writes)->toBe([]) + ->and(array_column($result['calls'], 0))->toContain('delivery'); + expect($result['native_helper_loaded'])->toBe($nativeCore); + // The parent must retain the held child outcome, regardless of launch form. + plugin_test_reset(); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 60]); + plugin_test_queue_db_result('exec_with_timeout', $exit); + plugin_test_queue_db_result('report_outcome', ['status' => 'running']); + global $config; + $oldBase = $config['base_path']; + $config['base_path'] = $this->fixtureRoot; + try { + expect($native ? flowview_reports_run(51) : reports_run(51))->toBeFalse(); + } finally { + $config['base_path'] = $oldBase; + } + + $mutations = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => str_starts_with($c['sql'], 'DELETE FROM reports_queued') || ($c['params'][0] ?? '') === 'pending')); + expect($mutations)->toBe([]); + + $released = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'unregister')); + expect($released)->toBe([['unregister', ['flowsched', 'child_7', 0, $result['pid']]]]); +})->with([false, true])->with(['notification-mail-error', 'notification-list-error', 'notification-log-error'])->with([false, true]); + +it('exposes lookup outcomes with the official native helper loaded', function ($mode) { + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/run_schedule.php', '--report-id=7', '--queue-id=51']; + [$exit] = ($this->runFixture)($command, $mode, true); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + $queueReads = array_filter($result['calls'], fn ($c) => $c[0] === 'row' && str_contains($c[1], 'reports_queued')); + $writes = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'write')); + expect($queueReads)->toHaveCount(2)->and($result['native_helper_loaded'])->toBeTrue(); + if ($mode === 'notification-success') { + expect($exit)->toBe(0)->and($writes)->toHaveCount(2) + ->and(end($writes)[2])->toBe(['complete', 51, 'flowview', 'running']); + } else { + expect($exit)->toBe(75)->and($writes)->toHaveCount(1) + ->and($writes[0][2])->toBe(['retry', 51, 'flowview', 'running']) + ->and(array_column($result['calls'], 0))->not->toContain('delivery', 'save-log'); + } +})->with(['notification-read-failure', 'notification-missing', 'notification-success']); + + +it('completes CC-only and BCC-only lists through the actual schedule child', function ($field, $nativeCore) { + $recipients = [$field => 'recipient@example.invalid']; + $command = [PHP_BINARY, $this->fixtureRoot . '/plugins/flowview/run_schedule.php', '--report-id=7', '--queue-id=51']; + [$exit] = ($this->runFixture)($command, 'notification-success', $nativeCore, $recipients); + $result = json_decode(file_get_contents($this->fixtureRoot . '/result.json'), true); + $writes = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'write')); + expect($exit)->toBe(0)->and($result['native_helper_loaded'])->toBe($nativeCore) + ->and($writes)->toHaveCount(2)->and($writes[0][1])->toContain('UPDATE plugin_flowview_schedules') + ->and($writes[1][2])->toBe(['complete', 51, 'flowview', 'running']); + $delivery = array_values(array_filter($result['calls'], fn ($c) => $c[0] === 'delivery')); + expect($delivery)->toHaveCount(1)->and($delivery[0][1])->toBe([ + '', $recipients['cc_emails'] ?? '', $recipients['bcc_emails'] ?? '' + ]); +})->with(['cc_emails', 'bcc_emails'])->with([false, true]); diff --git a/tests/Unit/FilterQueueBatchTest.php b/tests/Unit/FilterQueueBatchTest.php new file mode 100644 index 0000000..080df3c --- /dev/null +++ b/tests/Unit/FilterQueueBatchTest.php @@ -0,0 +1,247 @@ +oldBase = $config['base_path']; + $this->root = sys_get_temp_dir() . '/flowview-batch-' . bin2hex(random_bytes(8)); + mkdir($this->root . '/lib', 0700, true); + file_put_contents($this->root . '/lib/poller.php', 'root; +}); +afterEach(function () { + global $config; + $config['base_path'] = $this->oldBase; + unlink($this->root . '/lib/poller.php'); + rmdir($this->root . '/lib'); + rmdir($this->root); +}); +it('attempts a failed occurrence once and still dispatches the healthy row behind it', function () { + for ($i = 0;$i < 5;$i++) { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + } + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['id' => 51], ['id' => 52]]); + foreach ([51, 52] as $id) { + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => $id, 'source' => 'flowview', 'source_id' => $id - 44, 'run_command' => 'fixture', 'run_timeout' => 60]); + } + plugin_test_queue_db_result('exec_with_timeout', 75); + plugin_test_queue_db_result('report_outcome', ['status' => 'retry']); + plugin_test_queue_db_result('exec_with_timeout', 0); + plugin_test_queue_db_result('report_outcome', ['status' => 'complete']); + expect(flowview_reports_run_pending())->toBe(1); + $calls = $GLOBALS['__test_db_calls']; + $exec = array_values(array_filter($calls, fn ($c) => $c['fn'] === 'exec_with_timeout')); + expect($exec)->toHaveCount(2)->and($exec[0]['params'][0])->toContain('--queue-id=51')->and($exec[1]['params'][0])->toContain('--queue-id=52'); + $deletes = array_values(array_filter($calls, fn ($c) => str_starts_with($c['sql'], 'DELETE FROM reports_queued'))); + expect($deletes)->toHaveCount(1)->and($deletes[0]['params'])->toBe([52, 'flowview']); + $query = array_values(array_filter($calls, fn ($c) => $c['fn'] === 'db_fetch_assoc_prepared' && !str_starts_with($c['sql'], 'SELECT id,source_id')))[0]; + expect($query['sql'])->toContain('ORDER BY start_time,scheduled_time,id LIMIT 100'); +}); +it('stops the batch when schema readiness is lost instead of starting another child', function () { + foreach ([true, true, true, false] as $ready) { + plugin_test_queue_db_result('db_fetch_row', $ready ? ['Field' => 'enabled'] : []); + } + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['id' => 51], ['id' => 52]]); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 60]); + plugin_test_queue_db_result('exec_with_timeout', 75); + plugin_test_queue_db_result('report_outcome', ['status' => 'retry']); + expect(flowview_reports_run_pending())->toBe(0); + expect(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'exec_with_timeout'))->toHaveCount(1); +}); +it('does not dispatch beyond capacity or after a failed process count', function ($running) { + for ($i = 0;$i < 2;$i++) { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + } + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['id' => 51]]); + plugin_test_queue_db_result('db_fetch_assoc_prepared', $running === false ? false : array_fill(0, $running, ['taskname' => 'child_7', 'taskid' => 0, 'pid' => getmypid()])); + expect(flowview_reports_run_pending(2))->toBe(0); + expect(array_column($GLOBALS['__test_db_calls'], 'fn'))->not->toContain('exec_with_timeout'); +})->with([2, false]); +it('does not read a queue before schema readiness and does not ignore failed queue reads', function () { + expect(flowview_reports_run_pending())->toBe(0); + plugin_test_reset(); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_assoc_prepared', false); + expect(flowview_reports_run_pending(1, 100000))->toBe(0); + $query = end($GLOBALS['__test_db_calls']); + expect($query['sql'])->toContain('LIMIT 1000'); +}); + +it('reclaims a failed childs dead registration before dispatching the healthy occurrence', function () { + if (!function_exists('posix_kill')) { + $this->markTestSkipped('POSIX liveness is unavailable'); + } + for ($i = 0;$i < 5;$i++) { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + } + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['id' => 51], ['id' => 52]]); + plugin_test_queue_db_result('db_fetch_assoc_prepared', []); + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['taskname' => 'child_7', 'taskid' => 0, 'pid' => 2147483647]]); + foreach ([51, 52] as $id) { + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => $id, 'source' => 'flowview', 'source_id' => $id - 44, 'run_command' => 'fixture', 'run_timeout' => 60]); + } + plugin_test_queue_db_result('exec_with_timeout', 75); + plugin_test_queue_db_result('report_outcome', ['status' => 'retry']); + plugin_test_queue_db_result('exec_with_timeout', 0); + plugin_test_queue_db_result('report_outcome', ['status' => 'complete']); + expect(flowview_reports_run_pending())->toBe(1); + $calls = $GLOBALS['__test_db_calls']; + $deletes = array_values(array_filter($calls, fn ($c) => str_starts_with($c['sql'], 'DELETE FROM processes'))); + expect($deletes)->toHaveCount(1)->and($deletes[0]['params'])->toBe(['flowsched', 'child_7', 0, 2147483647]); + expect(array_filter($calls, fn ($c) => $c['fn'] === 'exec_with_timeout'))->toHaveCount(2); +}); +it('keeps live and uncertain child registrations and removes only confirmed dead ones', function () { + if (!function_exists('posix_kill')) { + $this->markTestSkipped('POSIX liveness is unavailable'); + } + plugin_test_queue_db_result('db_fetch_assoc_prepared', [ + ['taskname' => 'child_7', 'taskid' => 0, 'pid' => 2147483647], + ['taskname' => 'child_8', 'taskid' => 0, 'pid' => getmypid()], + ['taskname' => 'child_9', 'taskid' => 0, 'pid' => 0], + ]); + expect(flowview_running_schedule_children())->toBe(2); + expect(array_filter($GLOBALS['__test_db_calls'], fn ($c) => str_starts_with($c['sql'], 'DELETE')))->toHaveCount(1); + expect(flowview_schedule_pid_alive(getmypid()))->toBeTrue()->and(flowview_schedule_pid_alive(0))->toBeNull(); +}); +it('does not grant capacity when removal of a dead registration fails', function () { + if (!function_exists('posix_kill')) { + $this->markTestSkipped('POSIX liveness is unavailable'); + } + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['taskname' => 'child_7', 'taskid' => 0, 'pid' => 2147483647]]); + plugin_test_queue_db_result('db_execute_prepared', false); + $GLOBALS['__test_cacti_log'] = []; + expect(flowview_running_schedule_children())->toBe(1); + expect(implode(' ', $GLOBALS['__test_cacti_log']))->toContain('Unable to remove a dead FlowView child registration'); +}); + +it('keeps a capacity reservation when a concurrent replacement makes dead-PID deletion affect no rows', function () { + if (!function_exists('posix_kill')) { + $this->markTestSkipped('POSIX liveness is unavailable'); + } + for ($i = 0;$i < 2;$i++) { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + } + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['id' => 52]]); + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['taskname' => 'child_7', 'taskid' => 0, 'pid' => 2147483647]]); + plugin_test_queue_db_result('db_affected_rows', 0); + expect(flowview_reports_run_pending(1))->toBe(0); + expect(array_column($GLOBALS['__test_db_calls'], 'fn'))->not->toContain('exec_with_timeout'); + $deletes = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => str_starts_with($c['sql'], 'DELETE FROM processes'))); + expect($deletes)->toHaveCount(1)->and($deletes[0]['params'])->toBe(['flowsched', 'child_7', 0, 2147483647]); +}); + +it('does not launch or reset an occurrence when another dispatcher already owns its claim', function () { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 60]); + plugin_test_queue_db_result('db_affected_rows', 0); + expect(flowview_reports_run(51))->toBeFalse(); + $calls = $GLOBALS['__test_db_calls']; + expect(array_column($calls, 'fn'))->not->toContain('exec_with_timeout'); + $writes = array_values(array_filter($calls, fn ($c) => $c['fn'] === 'db_execute_prepared')); + expect($writes)->toHaveCount(1)->and($writes[0]['sql'])->toContain('AND status=?') + ->and($writes[0]['params'][4])->toBe('pending'); +}); + +it('retries successful completion cleanup without rerunning or requeuing the delivered report', function ($firstDelete, $retryDelete) { + $GLOBALS['__test_cacti_log'] = []; + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 60]); + plugin_test_queue_db_result('exec_with_timeout', 0); + plugin_test_queue_db_result('report_outcome', ['status' => 'complete']); + foreach ([true, $firstDelete, $retryDelete] as $result) { + plugin_test_queue_db_result('db_execute_prepared', $result); + } + expect(flowview_reports_run(51))->toBe($firstDelete || $retryDelete); + $calls = $GLOBALS['__test_db_calls']; + $exec = array_filter($calls, fn ($c) => $c['fn'] === 'exec_with_timeout'); + $writes = array_values(array_filter($calls, fn ($c) => $c['fn'] === 'db_execute_prepared')); + expect($exec)->toHaveCount(1)->and($writes)->toHaveCount($firstDelete ? 2 : 3); + foreach (array_slice($writes, 1) as $write) { + expect($write['sql'])->toBe('DELETE FROM reports_queued WHERE id=? AND source=?') + ->and($write['params'])->toBe([51, 'flowview']); + } + $log = implode(' ', $GLOBALS['__test_cacti_log']); + if (!$firstDelete && !$retryDelete) { + expect($log)->toContain('queue repair is required', '51', '7', 'already completed'); + } else { + expect($log)->not->toContain('queue repair is required'); + } +})->with([[true, true], [false, true], [false, false]]); + +it('does not requeue an unresolved timeout or crash and reconciles only explicit child outcomes', function ($outcome, $success, $reset, $status) { + $GLOBALS['__test_cacti_log'] = []; + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 1]); + plugin_test_queue_db_result('report_outcome', $outcome); + plugin_test_queue_db_result('exec_with_timeout', $status); + plugin_test_queue_db_result('exec_with_timeout_result', false); + expect(flowview_reports_run(51))->toBe($success); + $writes = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'db_execute_prepared')); + expect($writes)->toHaveCount(($success || $reset) ? 2 : 1); + if ($reset) { + expect($writes[1]['params'])->toBe(['pending', 51, 'flowview', 'retry']); + } elseif ($success) { + expect($writes[1]['sql'])->toStartWith('DELETE FROM reports_queued'); + } else { + expect(implode(' ', $GLOBALS['__test_cacti_log']))->toContain('unresolved child outcome', '51'); + } +})->with([[['status' => 'running'], false, false], [false, false, false], [[], false, false], [['status' => 'complete'], true, false], [['status' => 'retry'], false, true]])->with([0, 75, 1, -1, 255]); + +it('reconciles late complete and retry markers without losing queue ownership', function ($state, $reset, $claim, $success, $launches) { + plugin_test_queue_db_result('report_outcome', ['status' => 'complete']); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'status' => $state, 'run_command' => 'fixture', 'run_timeout' => 1]); + foreach ([$reset, $claim] as $value) { + plugin_test_queue_db_result('db_execute_prepared', $value); + } + expect(flowview_reports_run(51))->toBe($success); + expect(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'exec_with_timeout'))->toHaveCount($launches); +})->with([['complete', true, true, true, 0], ['retry', false, true, false, 0], ['retry', true, false, false, 0], ['retry', true, true, true, 1]]); + +it('cleans late completed reports even when execution capacity is full', function () { + for ($i = 0; $i < 3; $i++) { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + } + plugin_test_queue_db_result('completed_reports', [['id' => 51, 'source_id' => 7]]); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'status' => 'complete']); + expect(flowview_reports_run_pending())->toBe(1); + $exec = array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'exec_with_timeout'); + expect($exec)->toHaveCount(0); + $reads = array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'db_fetch_assoc_prepared' && str_contains($c['sql'], 'FROM processes')); + expect($reads)->toHaveCount(0); +}); + +it('preserves legacy dispatch for other report sources', function () { + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'other-plugin', 'source_id' => 7, 'name' => 'Other report', 'run_command' => 'fixture', 'run_timeout' => 60]); + expect(reports_run(51))->toBeTrue(); + $exec = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'exec_with_timeout')); + expect($exec)->toHaveCount(1)->and($exec[0]['params'])->toBe(['fixture --report-id=51', 60]); +}); + +it('uses the scheduler default when a report timeout is unset', function () { + plugin_test_queue_db_result('report_outcome', ['status' => 'complete']); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 0]); + expect(flowview_reports_run(51))->toBeTrue(); + $exec = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'exec_with_timeout')); + expect($exec[0]['params'][1])->toBe(300); +}); + +it('continues executable work after a cleanup snapshot read fails', function () { + $GLOBALS['__test_cacti_log'] = []; + for ($i = 0; $i < 3; $i++) { plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); } + plugin_test_queue_db_result('completed_reports', false); + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['id' => 51]]); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 1]); + plugin_test_queue_db_result('report_outcome', ['status' => 'complete']); + expect(flowview_reports_run_pending())->toBe(1); + expect(implode(' ', $GLOBALS['__test_cacti_log']))->toContain('Unable to read completed FlowView occurrences'); +}); + +it('retains its completed cleanup count when the executable snapshot read fails', function () { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('completed_reports', [['id' => 51, 'source_id' => 7]]); + plugin_test_queue_db_result('db_fetch_assoc_prepared', false); + expect(flowview_reports_run_pending())->toBe(1); + expect(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'exec_with_timeout'))->toHaveCount(0); +}); diff --git a/tests/Unit/FilterScheduleRetryTest.php b/tests/Unit/FilterScheduleRetryTest.php new file mode 100644 index 0000000..f746e66 --- /dev/null +++ b/tests/Unit/FilterScheduleRetryTest.php @@ -0,0 +1,153 @@ +toBe($expected); +})->with([ + [['report-id' => 7, 'queue-id' => 51], 51], [['report-id' => 51], 51], + [['report-id' => 7, 'queue-id' => false], 7], [['report-id' => 0], false], + [['report-id' => 7, 'queue-id' => 'invalid'], false], [[], false], +]); +it('keeps a pending occurrence untouched before either runner claims it', function ($legacy) { + if ($legacy) { + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview']); + } + expect($legacy ? reports_run(51) : flowview_reports_run(51))->toBeFalse(); + expect(array_column($GLOBALS['__test_db_calls'], 'fn'))->not->toContain('db_execute_prepared', 'exec_with_timeout'); +})->with([false, true]); +it('retains the original on deferred or failed children and only deletes successful occurrences', function ($status, $reset, $legacy, $startup, $warning = false) { + global $config; + $old = $config['base_path']; + $root = sys_get_temp_dir() . '/flowview-shim-' . bin2hex(random_bytes(8)); + mkdir($root . '/lib', 0700, true); + file_put_contents($root . '/lib/poller.php', ' 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'name' => 'Fixture', 'source_id' => 7, 'run_command' => 'fixture-command', 'run_timeout' => $status === 1 ? 0 : 60]); + plugin_test_queue_db_result('exec_with_timeout', $startup ? null : $status); + plugin_test_queue_db_result('report_outcome', ['status' => $status === 0 ? 'complete' : 'retry']); + plugin_test_queue_db_result('exec_with_timeout_result', ($startup || $warning) ? false : null); + plugin_test_queue_db_result('db_execute_prepared', true); + plugin_test_queue_db_result('db_execute_prepared', $reset); + $success = $status === 0 && !$startup && $reset; + expect($legacy ? reports_run(51) : flowview_reports_run(51))->toBe($success); + $calls = $GLOBALS['__test_db_calls']; + $sql = implode(' ', array_column($calls, 'sql')); + if ($status !== 0 || $startup) { + if (!$reset) { + expect(implode(' ', $GLOBALS['__test_cacti_log']))->toContain('queue repair is required'); + } + expect($sql)->not->toContain('DELETE FROM reports_queued'); + expect($sql)->toContain('UPDATE reports_queued SET status'); + } else { + expect($sql)->toContain('DELETE FROM reports_queued'); + $writes = array_values(array_filter($calls, fn ($c) => $c['fn'] === 'db_execute_prepared')); + expect($writes)->toHaveCount(2)->and($writes[1]['sql'])->toStartWith('DELETE FROM reports_queued'); + } + if (!$legacy) { + $exec = array_values(array_filter($calls, fn ($c) => $c['fn'] === 'exec_with_timeout'))[0]; + expect($exec['params'][0])->toContain('--report-id=7 --queue-id=51'); + expect($exec['params'][1])->toBe($status === 1 ? 300 : 60); + } + expect($GLOBALS['__test_saved_rows'])->toBe([]); + } finally { + $config['base_path'] = $old; + unlink($root . '/lib/poller.php'); + rmdir($root . '/lib'); + rmdir($root); + } +})->with([[75, true, false, false], [75, false, false, false], [0, true, false, true], [0, true, false, false], [75, true, true, false], [75, false, true, false], [0, true, true, false], [0, true, true, true], [0, true, false, false, true], [0, true, true, false, true]]); +it('rejects missing queue rows or failed claims without running a child', function ($missing) { + global $config; + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', $missing ? [] : ['id' => 51, 'source' => 'flowview']); + if (!$missing) { + // A previously included harmless fixture path satisfies require_once. + $old = $config['base_path']; + $root = sys_get_temp_dir() . '/flowview-claim-' . bin2hex(random_bytes(8)); + mkdir($root . '/lib', 0700, true); + file_put_contents($root . '/lib/poller.php', 'toBeFalse(); + expect(array_column($GLOBALS['__test_db_calls'], 'fn'))->not->toContain('exec_with_timeout'); + } finally { + if (!$missing) { + $config['base_path'] = $old; + unlink($root . '/lib/poller.php'); + rmdir($root . '/lib'); + rmdir($root); + } + } +})->with([true, false]); + +it('runs the actual child entrypoint with native and legacy dispatch IDs', function ($native, $ready) { + $root = sys_get_temp_dir() . '/flowview-entrypoint-' . bin2hex(random_bytes(8)); + mkdir($root . '/plugins/flowview/includes', 0700, true); + mkdir($root . '/include'); + mkdir($root . '/lib'); + $plugin = dirname(__DIR__, 2); + copy($plugin . '/run_schedule.php', $root . '/plugins/flowview/run_schedule.php'); + foreach (['database.php', 'functions.php'] as $file) { + symlink($plugin . '/includes/' . $file, $root . '/plugins/flowview/includes/' . $file); + } + file_put_contents($root . '/plugins/flowview/setup.php', '"enabled"]);plugin_test_queue_db_result("register_process_start", false);' : ''; + file_put_contents($root . '/include/cli_check.php', '51,"source"=>"flowview","source_id"=>7]); register_shutdown_function(function(){register_shutdown_function(function(){file_put_contents(' . $result . ',json_encode($GLOBALS["__test_db_calls"]));});});'); + try { + $command = [PHP_BINARY, $root . '/plugins/flowview/run_schedule.php', '--report-id=' . ($native ? 7 : 51)]; + if ($native) { + $command[] = '--queue-id=51'; + } + $process = proc_open($command, [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes); + fclose($pipes[0]); + $output = stream_get_contents($pipes[1]); + $error = stream_get_contents($pipes[2]); + fclose($pipes[1]); + fclose($pipes[2]); + $exit = proc_close($process); + expect($error)->toBe('')->and($output)->toBe('')->and($exit)->toBe(75); + $calls = json_decode(file_get_contents($root . '/result.json'), true); + expect($calls)->toHaveCount($ready ? 4 : 3)->and($calls[0]['params'])->toBe(['flowview', 51]); + $markers = array_values(array_filter($calls, fn ($c) => $c['fn'] === 'db_execute_prepared')); + expect($markers)->toHaveCount(1)->and($markers[0]['params'])->toBe(['retry', 51, 'flowview', 'running']); + if ($ready) { + global $config; + $oldBase = $config['base_path']; + $config['base_path'] = $root; + try { + plugin_test_reset(); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'run_command' => 'fixture', 'run_timeout' => 60]); + plugin_test_queue_db_result('exec_with_timeout', $exit); + plugin_test_queue_db_result('report_outcome', ['status' => 'retry']); + expect(flowview_reports_run(51))->toBeFalse(); + expect(implode(' ', array_column($GLOBALS['__test_db_calls'], 'sql')))->not->toContain('DELETE FROM reports_queued'); + } finally { + $config['base_path'] = $oldBase; + } + } + } finally { + $iterator = new RecursiveIteratorIterator(new RecursiveDirectoryIterator($root, FilesystemIterator::SKIP_DOTS), RecursiveIteratorIterator::CHILD_FIRST); + foreach ($iterator as $file) { + if ($file->isDir() && !$file->isLink()) { + rmdir($file->getPathname()); + } else { + unlink($file->getPathname()); + } + } + rmdir($root); + } +})->with([[false, false], [true, false], [false, true], [true, true]]); diff --git a/tests/Unit/FilterStatusTest.php b/tests/Unit/FilterStatusTest.php new file mode 100644 index 0000000..6752d17 --- /dev/null +++ b/tests/Unit/FilterStatusTest.php @@ -0,0 +1,266 @@ + 'on']))->toBe(['class' => 'deviceUp', 'label' => 'Enabled']) + ->and(flowview_filter_status(['enabled' => '']))->toBe(['class' => 'deviceDisabled', 'label' => 'Disabled']) + ->and(flowview_filter_status([]))->toBe(['class' => 'deviceDisabled', 'label' => 'Unavailable']); +}); + +it('does not alter a missing filter table', function () { + expect(flowview_filter_status_upgrade())->toBeFalse(); + expect(array_column($GLOBALS['__test_db_calls'], 'fn'))->not->toContain('db_execute'); +}); + +it('leaves an existing enabled column intact', function () { + plugin_test_queue_db_result('db_fetch_cell', 1); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + expect(flowview_filter_status_upgrade())->toBeTrue(); + expect(array_column($GLOBALS['__test_db_calls'], 'fn'))->not->toContain('db_execute'); +}); + +it('repairs legacy filters as enabled through the configured database', function () { + plugin_test_queue_db_result('db_fetch_cell', 1); + expect(flowview_filter_status_upgrade())->toBeTrue(); + $writes = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($call) => $call['fn'] === 'db_execute')); + expect($writes)->toHaveCount(1) + ->and($writes[0]['sql'])->toBe("ALTER TABLE plugin_flowview_queries ADD COLUMN enabled char(2) NOT NULL DEFAULT 'on' AFTER name") + ->and($writes[0]['conn'])->toBe('filter-database'); +}); + +it('rechecks concurrent upgrades and reports actual failures', function ($exists) { + plugin_test_queue_db_result('db_fetch_cell', 1); + plugin_test_queue_db_result('db_fetch_row', []); + plugin_test_queue_db_result('db_fetch_row', $exists ? ['Field' => 'enabled'] : []); + plugin_test_queue_db_result('db_execute', false); + expect(flowview_filter_status_upgrade())->toBe($exists); +})->with([true, false]); + +it('does not run a schedule when the state schema is unavailable', function () { + expect(plugin_flowview_run_schedule(1, 1))->toBeFalse(); + expect(array_column($GLOBALS['__test_db_calls'], 'fn'))->not->toContain('db_fetch_row_prepared'); +}); + +it('does not prepare scheduled reports for missing or disabled filters', function ($query) { + plugin_test_queue_db_result('db_fetch_cell', 1); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['query_id' => 7]); + plugin_test_queue_db_result('db_fetch_row_prepared', $query); + expect(plugin_flowview_run_schedule(1, 1))->toBeTrue(); + $reads = array_filter($GLOBALS['__test_db_calls'], fn ($call) => $call['fn'] === 'db_fetch_row_prepared'); + expect($reads)->toHaveCount(2); +})->with([[[]], [['enabled' => '']], [['name' => 'legacy']]]); + +it('continues preparing an enabled scheduled filter', function () { + plugin_test_queue_db_result('db_fetch_cell', 1); + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['query_id' => 7]); + plugin_test_queue_db_result('db_fetch_row_prepared', ['enabled' => 'on', 'timespan' => 1]); + plugin_test_queue_db_result('get_timespan', new RuntimeException('enabled report reached time selection')); + expect(fn () => plugin_flowview_run_schedule(1, 1))->toThrow(RuntimeException::class, 'enabled report reached time selection'); +}); + +it('preserves an existing disabled schedule filter with an explicit label', function () { + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['id' => 2, 'name' => 'Active', 'enabled' => 'on'], ['id' => 7, 'name' => 'Saved', 'enabled' => '']]); + expect(flowview_schedule_filters(3))->toBe([2 => 'Active', 7 => 'Saved (Disabled)']); + $call = $GLOBALS['__test_db_calls'][0]; + expect($call['params'])->toBe([3])->and($call['sql'])->toContain('SELECT query_id'); +}); +it('handles unavailable schedule choices without replacing the association', function () { + plugin_test_queue_db_result('db_fetch_assoc_prepared', false); + expect(flowview_schedule_filters(3))->toBe([]); +}); + +it('resets a queued time-selection exception between tests', function () { + plugin_test_queue_db_result('get_timespan', new RuntimeException('fixture only')); + plugin_test_reset(); + $span = []; + get_timespan($span, 2000, 1, 0); + expect($span)->toBe(['begin_now' => 200, 'end_now' => 2000]); +}); + +it('distinguishes prepared records, an empty period and a failed query', function ($data, $success, $notificationReport = 'valid') { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['query_id' => 3, 'title' => 'Fixture', 'format_file' => 'default']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['enabled' => 'on', 'timespan' => 1, 'name' => 'Fixture']); + plugin_test_queue_db_result('get_timespan', ['begin_now' => 0, 'end_now' => 2000]); + set_request_var('timespan', 'custom'); + set_request_var('query', 3); + set_request_var('date1', date('Y-m-d H:i:s', 100)); + set_request_var('date2', date('Y-m-d H:i:s', 2000)); + $key = get_flowview_session_key(3, 100, 2000); + $_SESSION['sess_flowdata'][$key] = ['timeout' => time() + 60, 'data' => $data]; + plugin_test_queue_db_result('db_fetch_row_prepared', $notificationReport === 'valid' ? ['name' => 'Fixture', 'notification' => '', 'request_type' => 'scheduled', 'requested_by' => 'fixture', 'requested_id' => 1] : $notificationReport); + plugin_test_queue_db_result('sql_save', 1); + $GLOBALS['__test_cacti_log'] = []; + expect(plugin_flowview_run_schedule(7, 51))->toBe($success); + $formatted = in_array('reports_load_format_file', array_column($GLOBALS['__test_db_calls'], 'fn'), true); + expect($formatted)->toBe(!empty($data['data'])); + if ($data === false || ($data['data'] ?? null) === false) { + expect(implode(' ', $GLOBALS['__test_cacti_log']))->toContain('retry'); + } elseif (empty($data['data'])) { + expect(implode(' ', $GLOBALS['__test_cacti_log']))->toContain('Returned not Data'); + } +})->with([ + [['table' => 'Fixture', 'data' => [['bytes' => 42]]], true], + [['table' => 'Fixture', 'data' => [['bytes' => 42]]], false, false], + [['table' => 'Fixture', 'data' => [['bytes' => 42]]], false, []], + [['table' => 'Fixture', 'data' => []], true], + [false, false], + [['table' => 'Fixture', 'data' => false], false], +]); + +it('retries failed filter lookups while consuming genuinely absent or disabled filters', function ($filter, $success) { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['query_id' => 3]); + plugin_test_queue_db_result('db_fetch_row_prepared', $filter); + expect(plugin_flowview_run_schedule(7, 51))->toBe($success); +})->with([[false, false], [[], true], [['enabled' => ''], true]]); + +it('distinguishes uncached empty query execution from failure in scheduled preparation', function ($rows, $success, $catalog = 'normal', $details = 'normal', $threads = 1, $parallelResult = '[]') { + $start = strtotime('2024-01-01 00:00:00 UTC'); + $end = $start + 3600; + $query = ['id' => 3, 'name' => 'Fixture', 'enabled' => 'on', 'timespan' => 1, 'statistics' => 9, 'printed' => 0, 'sortfield' => 'bytes', 'cutofflines' => 100]; + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row', $details === 'normal' ? ['min_date' => date('Y-m-d H:i:s', $start), 'max_date' => date('Y-m-d H:i:s', $end), 'table_partition' => ''] : $details); + plugin_test_queue_db_result('db_fetch_row_prepared', ['query_id' => 3, 'title' => 'Fixture', 'format_file' => 'default']); + plugin_test_queue_db_result('db_fetch_row_prepared', $query); + plugin_test_queue_db_result('db_fetch_row_prepared', $query); + plugin_test_queue_db_result('get_timespan', ['begin_now' => $start, 'end_now' => $end]); + // arrays.php reads three choice lists each time; no result-session cache is seeded. + for ($i = 0; $i < 6; $i++) { + plugin_test_queue_db_result('db_fetch_assoc', []); + } + plugin_test_queue_db_result('db_fetch_assoc', $catalog === 'normal' ? [['table' => 'plugin_flowview_raw_202400000']] : $catalog); + plugin_test_queue_db_result('db_fetch_assoc_prepared', $rows); + $GLOBALS['__test_config']['flowview_partition'] = 1; + $GLOBALS['__test_config']['flowview_parallel_threads'] = $threads; + if ($threads > 1) { + $GLOBALS['debug'] = false; + $GLOBALS['__test_config']['flowview_parallel_runlimit'] = 5; + foreach (['Fixture', 71, 0, $parallelResult, ''] as $value) { + plugin_test_queue_db_result('db_fetch_cell_prepared', $value); + } + plugin_test_queue_db_result('db_fetch_cell', 0); + } + expect(plugin_flowview_run_schedule(7, 51))->toBe($success); + $queries = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'db_fetch_assoc_prepared')); + if ($threads > 1) { + $reads = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'db_fetch_cell_prepared' && str_contains($c['sql'], 'SELECT results'))); + expect($reads)->toHaveCount(1)->and($reads[0]['params'])->toBe([71]); + } elseif ($catalog === 'normal' && $details === 'normal') { + expect($queries)->toHaveCount(1)->and($queries[0]['sql'])->toContain('FROM plugin_flowview_raw_202400000'); + } else { + expect($queries)->toHaveCount(0); + expect($_SESSION['sess_flowview_table_details'] ?? [])->not->toHaveKey('plugin_flowview_raw_202400000'); + } + expect($_SESSION)->not->toHaveKey('sess_flowdata'); +})->with([ + [[], true], [false, false], + [[], false, false], [[], true, []], + [[], false, 'normal', false], [[], false, 'normal', []], + [[], true, 'normal', ['min_date' => null, 'max_date' => null, 'table_partition' => '']], + [[], true, 'normal', 'normal', 2, '[]'], + [[], false, 'normal', 'normal', 2, false], + [[], false, 'normal', 'normal', 2, 'null'], + [[], false, 'normal', 'normal', 2, 'invalid-json'], +]); + +it('distinguishes failed and absent schedule lookups before reading a filter', function ($schedule, $success) { + plugin_test_queue_db_result('db_fetch_row', ['Field' => 'enabled']); + plugin_test_queue_db_result('db_fetch_row_prepared', $schedule); + expect(plugin_flowview_run_schedule(7, 51))->toBe($success); + $reads = array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'db_fetch_row_prepared'); + expect($reads)->toHaveCount(1); +})->with([[false, false], [[], true]]); +it('fails an uncached query cleanly when its own filter lookup fails or is absent', function ($filter) { + plugin_test_queue_db_result('db_fetch_row_prepared', $filter); + expect(run_flow_query(false, 3, 1704067200, 1704070800))->toBeFalse(); +})->with([[false], [[]]]); + +it('preserves two-request reduce lookup failures and cleans up both requests', function ($reduce, $success, $metadata = 'normal') { + $GLOBALS['debug'] = false; + $GLOBALS['__test_config']['flowview_parallel_runlimit'] = 5; + $query = ['sql_query' => 'SELECT SUM(bytes) bytes', 'sql_where' => '', 'sql_groupby' => '', 'sql_having' => '', 'sql_order' => '', 'sql_limit' => '', 'sql_params' => []]; + foreach ([0, 0, $reduce === 'valid' ? json_encode($query) : $reduce, '', ''] as $value) { + plugin_test_queue_db_result('db_fetch_cell_prepared', $value); + } + plugin_test_queue_db_result('db_fetch_cell', 0); + plugin_test_queue_db_result('db_fetch_assoc', $metadata === 'normal' ? [ + ['map_table' => 'fixture_map_71', 'map_create' => 'CREATE TABLE fixture_map_71 (bytes bigint)', 'results' => '[]'], + ['map_table' => 'fixture_map_72', 'map_create' => 'CREATE TABLE fixture_map_72 (bytes bigint)', 'results' => '[]'], + ] : $metadata); + expect(parallel_database_query_run([71, 72]))->toBe($success ? [] : false); + $cleanup = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => str_starts_with($c['sql'], 'DELETE FROM parallel_database_query_shard'))); + expect(array_column($cleanup, 'params'))->toBe([[71], [72]]); + $removed = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => preg_match('/^DELETE FROM parallel_database_query\s+WHERE/', $c['sql']))); + expect(array_column($removed, 'params'))->toBe($success ? [] : [[71], [72]]); +})->with([[false, false], ['', false], ['null', false], ['bad-json', false], ['valid', true], ['valid', false, false], ['valid', false, []], ['{}', false]]); + +it('returns a failed result after cleaning up timed out parallel requests', function () { + $GLOBALS['debug'] = false; + $GLOBALS['__test_config']['flowview_parallel_runlimit'] = 0; + foreach ([0, 0, '', ''] as $value) { + plugin_test_queue_db_result('db_fetch_cell_prepared', $value); + } + expect(parallel_database_query_run([71, 72]))->toBeFalse(); + $cleanup = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => str_starts_with($c['sql'], 'DELETE FROM parallel_database_query_shard'))); + expect(array_column($cleanup, 'params'))->toBe([[71], [72]]); +}); + +it('rejects a failed component before reducing any two-request results', function ($failed, $position) { + $GLOBALS['debug'] = false; + $GLOBALS['__test_config']['flowview_parallel_runlimit'] = 5; + $query = ['sql_query' => 'SELECT SUM(bytes) bytes', 'sql_where' => '', 'sql_groupby' => '', 'sql_having' => '', 'sql_order' => '', 'sql_limit' => '', 'sql_params' => []]; + foreach ([0, 0, json_encode($query), '', ''] as $value) { + plugin_test_queue_db_result('db_fetch_cell_prepared', $value); + } + plugin_test_queue_db_result('db_fetch_cell', 0); + $tables = [ + ['map_table' => 'fixture_map_71', 'map_create' => 'CREATE TABLE fixture_map_71 (bytes bigint)', 'results' => '[{"bytes":123}]'], + ['map_table' => 'fixture_map_72', 'map_create' => 'CREATE TABLE fixture_map_72 (bytes bigint)', 'results' => '[{"bytes":456}]'], + ]; + $tables[$position]['results'] = $failed; + plugin_test_queue_db_result('db_fetch_assoc', $tables); + expect(parallel_database_query_run([71, 72]))->toBeFalse(); + $cleanup = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => str_starts_with($c['sql'], 'DELETE FROM parallel_database_query_shard'))); + expect(array_column($cleanup, 'params'))->toBe([[71], [72]]); + $removed = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => preg_match('/^DELETE FROM parallel_database_query\s+WHERE/', $c['sql']))); + expect(array_column($removed, 'params'))->toBe([[71], [72]]); + $reductions = array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'db_execute' || ($c['fn'] === 'db_fetch_assoc_prepared' && str_contains($c['sql'], 'SUM(bytes)'))); + expect($reductions)->toBe([]); +})->with(['false', 'null', 'invalid-json'])->with([0, 1]); + +it('fails two-request materialization rather than reducing partial data', function ($writes, $success, $insertCount) { + $GLOBALS['debug'] = false; + $GLOBALS['__test_config']['flowview_parallel_runlimit'] = 5; + $query = ['sql_query' => 'SELECT SUM(bytes) bytes', 'sql_where' => '', 'sql_groupby' => '', 'sql_having' => '', 'sql_order' => '', 'sql_limit' => '', 'sql_params' => []]; + foreach ([0, 0, json_encode($query), '', ''] as $value) { + plugin_test_queue_db_result('db_fetch_cell_prepared', $value); + } + plugin_test_queue_db_result('db_fetch_cell', 0); + plugin_test_queue_db_result('db_fetch_assoc', [ + ['map_table' => 'fixture_map_71', 'map_create' => 'CREATE TABLE fixture_map_71 (bytes bigint)', 'results' => '[{"bytes":123}]'], + ['map_table' => 'fixture_map_72', 'map_create' => 'CREATE TABLE fixture_map_72 (bytes bigint)', 'results' => '[{"bytes":456}]'], + ]); + foreach ($writes as $result) { + plugin_test_queue_db_result('db_execute', $result); + } + plugin_test_queue_db_result('db_fetch_assoc_prepared', [['bytes' => 579]]); + expect(parallel_database_query_run([71, 72]))->toBe($success ? [['bytes' => 579]] : false); + $calls = $GLOBALS['__test_db_calls']; + $inserts = array_filter($calls, fn ($c) => $c['fn'] === 'db_execute' && str_starts_with($c['sql'], 'INSERT INTO')); + $reductions = array_filter($calls, fn ($c) => $c['fn'] === 'db_fetch_assoc_prepared' && str_contains($c['sql'], 'SUM(bytes)')); + $drops = array_filter($calls, fn ($c) => $c['fn'] === 'db_execute' && str_starts_with($c['sql'], 'DROP TEMPORARY TABLE IF EXISTS')); + $removed = array_values(array_filter($calls, fn ($c) => preg_match('/^DELETE FROM parallel_database_query\s+WHERE/', $c['sql']))); + expect($inserts)->toHaveCount($insertCount)->and($reductions)->toHaveCount($success ? 1 : 0) + ->and($drops)->toHaveCount(1)->and(array_column($removed, 'params'))->toBe($success ? [] : [[71], [72]]); +})->with([[[false], false, 0], [[true, false], false, 1], [[true, true, false], false, 2], [[true, true, true], true, 2]]); diff --git a/tests/Unit/ReportDeliveryTest.php b/tests/Unit/ReportDeliveryTest.php new file mode 100644 index 0000000..18fe914 --- /dev/null +++ b/tests/Unit/ReportDeliveryTest.php @@ -0,0 +1,96 @@ + 'Fixture', 'notification' => $payload, 'request_type' => 'scheduled', + 'requested_by' => 'fixture', 'requested_id' => 1 + ]); + plugin_test_queue_db_result('db_fetch_row_prepared', $list); + plugin_test_queue_db_result('mailer', $mailError); + plugin_test_queue_db_result('sql_save', $saved); + plugin_test_queue_db_result('delivery_format_ok', $format); + plugin_test_queue_db_result('delivery_format_tag', $tag); + $data = [['bytes' => 42]]; + $raw = '

Fixture

'; + $html = 'Fixture'; + $text = null; + expect(flowview_reports_log_and_notify(51, microtime(true), 'html', 'flowview', 7, 'Fixture', $data, $raw, $html, $text))->toBe($expected); + $mail = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'mailer')); + if ($expected === true) { + expect($GLOBALS['__test_saved_rows'])->toHaveCount(1) + ->and($GLOBALS['__test_saved_rows'][0]['table'])->toBe('reports_log'); + } elseif ($mailError !== '' || $saved === false) { + expect($mail)->toHaveCount(1); + } else { + expect($mail)->toBe([]); + } +})->with([ + ['', [], '', 1, true], + ['{invalid', [], '', 1, null], + ['null', [], '', 1, null], + ['{"email":false}', [], '', 1, null], + ['{"email":{}}', [], '', 1, null], + ['{"unknown":{}}', [], '', 1, null], + ['{"notification_list":{}}', [], '', 1, null], + ['{"notification_list":{"id":1}}', false, '', 1, null], + ['{"notification_list":{"id":1}}', [], '', 1, null], + ['{"notification_list":{"id":1}}', ['emails' => ''], '', 1, null], + ['{"email":{"to_email":"to@example.invalid"}}', [], '', 1, true], + ['{"email":{"to_email":"to@example.invalid","cc_email":"cc@example.invalid","bcc_email":"bcc@example.invalid","reply_to":"reply@example.invalid","from":{"email":"from@example.invalid","name":"Fixture"}}}', [], '', 1, true], + ['{"email":{"to_email":"to@example.invalid"}}', [], 'SMTP failed', 1, null], + ['{"email":{"to_email":"to@example.invalid"}}', [], '', false, null], + ['{"notification_list":{"id":1}}', ['emails' => 'to@example.invalid', 'format_file' => ''], '', 1, true], + ['{"notification_list":{"id":1,"reply_to":"fallback@example.invalid"}}', ['emails' => 'to@example.invalid', 'format_file' => 'default', 'cc_emails' => 'cc@example.invalid', 'bcc_emails' => 'bcc@example.invalid', 'reply_to' => 'reply@example.invalid'], '', 1, true, true, false], + ['{"notification_list":{"id":1}}', ['emails' => 'to@example.invalid', 'format_file' => 'default'], '', 1, true, false], + ['{"notification_list":{"id":1}}', ['emails' => 'to@example.invalid', 'format_file' => 'default'], 'SMTP failed', 1, null], +]); + +it('keeps a partial multi-notification delivery unresolved without writing a success log', function () { + plugin_test_queue_db_result('db_fetch_row_prepared', ['notification' => '{"email":{"to_email":"to@example.invalid"},"notification_list":{"id":1}}']); + plugin_test_queue_db_result('db_fetch_row_prepared', ['emails' => 'list@example.invalid', 'format_file' => 'default']); + plugin_test_queue_db_result('mailer', ''); + plugin_test_queue_db_result('mailer', 'SMTP failed after first notification'); + $data = []; $raw = ''; $html = ''; $text = ''; + expect(flowview_reports_log_and_notify(51, microtime(true), 'html', 'flowview', 7, 'Fixture', $data, $raw, $html, $text))->toBeNull(); + $mail = array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'mailer'); + expect($mail)->toHaveCount(2)->and($GLOBALS['__test_saved_rows'])->toBe([]); +}); + + +it('preserves the compatibility entrypoint outcome without hiding lookup failures', function ($report, $expected) { + plugin_test_queue_db_result('db_fetch_row_prepared', $report); + plugin_test_queue_db_result('sql_save', 1); + $data = []; $raw = ''; $html = ''; $text = ''; + expect(reports_log_and_notify(51, microtime(true), 'html', 'flowview', 7, 'Fixture', $data, $raw, $html, $text))->toBe($expected); +})->with([ + [false, false], [[], false], + [['notification' => '', 'name' => 'Fixture', 'request_type' => 'scheduled', 'requested_by' => 'fixture', 'requested_id' => 1], true] +]); + + +it('preserves every supported To CC and BCC recipient combination', function ($type, $mask, $mailError) { + $to = ($mask & 1) ? 'to@example.invalid' : ''; + $cc = ($mask & 2) ? 'cc@example.invalid' : ''; + $bcc = ($mask & 4) ? 'bcc@example.invalid' : ''; + $notification = $type === 'email' + ? ['email' => ['to_email' => $to, 'cc_email' => $cc, 'bcc_email' => $bcc]] + : ['notification_list' => ['id' => 1]]; + plugin_test_queue_db_result('db_fetch_row_prepared', [ + 'name' => 'Fixture', 'notification' => json_encode($notification), 'request_type' => 'scheduled', + 'requested_by' => 'fixture', 'requested_id' => 1 + ]); + plugin_test_queue_db_result('db_fetch_row_prepared', ['emails' => $to, 'cc_emails' => $cc, 'bcc_emails' => $bcc, 'format_file' => 'default']); + plugin_test_queue_db_result('mailer', $mailError); + plugin_test_queue_db_result('sql_save', 1); + $data = []; $raw = ''; $html = ''; $text = ''; + expect(flowview_reports_log_and_notify(51, microtime(true), 'html', 'flowview', 7, 'Fixture', $data, $raw, $html, $text))->toBe($mask && $mailError === '' ? true : null); + $mail = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'mailer')); + if ($mask) { + expect($mail)->toHaveCount(1)->and(array_slice($mail[0]['params'], 1, 3))->toBe([$to, $cc, $bcc]); + } else { + expect($mail)->toBe([]); + } + expect($GLOBALS['__test_saved_rows'])->toHaveCount($mask && $mailError === '' ? 1 : 0); +})->with(['email', 'notification_list'])->with(range(0, 7))->with(['', 'SMTP failed']); diff --git a/tests/bootstrap-unit.php b/tests/bootstrap-unit.php index c83ff97..77bb8d8 100644 --- a/tests/bootstrap-unit.php +++ b/tests/bootstrap-unit.php @@ -95,6 +95,8 @@ function plugin_test_reset() { $GLOBALS['__test_db_results'] = array(); $GLOBALS['__test_config'] = array(); $GLOBALS['__test_messages'] = array(); + $GLOBALS['__test_saved_rows'] = []; + $GLOBALS['__test_user_settings'] = []; $_GET = array(); $_POST = array(); $_REQUEST = array(); @@ -120,17 +122,23 @@ function plugin_test_db_result($function, $default) { } } +if (!function_exists('db_qstr')) { + function db_qstr($value) { + return "'" . str_replace("'", "''", (string) $value) . "'"; + } +} + if (!function_exists('db_execute')) { function db_execute($sql, $log = true, $conn = false) { $GLOBALS['__test_db_calls'][] = array('fn' => 'db_execute', 'sql' => $sql, 'params' => array(), 'conn' => $conn); - return true; + return plugin_test_db_result('db_execute', true); } } if (!function_exists('db_execute_prepared')) { function db_execute_prepared($sql, $params = array(), $log = true, $conn = false) { $GLOBALS['__test_db_calls'][] = array('fn' => 'db_execute_prepared', 'sql' => $sql, 'params' => $params, 'conn' => $conn); - return true; + return plugin_test_db_result('db_execute_prepared', true); } } @@ -144,6 +152,9 @@ function db_fetch_assoc($sql, $log = true, $conn = false) { if (!function_exists('db_fetch_assoc_prepared')) { function db_fetch_assoc_prepared($sql, $params = array(), $log = true, $conn = false) { $GLOBALS['__test_db_calls'][] = array('fn' => 'db_fetch_assoc_prepared', 'sql' => $sql, 'params' => $params, 'conn' => $conn); + if (str_starts_with($sql, 'SELECT id,source_id FROM reports_queued')) { + return plugin_test_db_result('completed_reports', []); + } return plugin_test_db_result('db_fetch_assoc_prepared', array()); } } @@ -158,6 +169,9 @@ function db_fetch_row($sql, $log = true, $conn = false) { if (!function_exists('db_fetch_row_prepared')) { function db_fetch_row_prepared($sql, $params = array(), $log = true, $conn = false) { $GLOBALS['__test_db_calls'][] = array('fn' => 'db_fetch_row_prepared', 'sql' => $sql, 'params' => $params, 'conn' => $conn); + if (str_starts_with($sql, 'SELECT status FROM reports_queued')) { + return plugin_test_db_result('report_outcome', []); + } return plugin_test_db_result('db_fetch_row_prepared', array()); } } @@ -201,7 +215,7 @@ function db_index_exists($table, $index) { if (!function_exists('db_column_exists')) { function db_column_exists($table, $column) { - return false; + return plugin_test_db_result('db_column_exists', false); } } @@ -303,6 +317,12 @@ function is_realm_allowed($realm) { } } +if (!function_exists('raise_message_javascript')) { + function raise_message_javascript(...$args) { + raise_message(...$args); + } +} + if (!function_exists('raise_message')) { function raise_message($id, $text = '', $level = 0) { $GLOBALS['__test_messages'][] = array($id, $text, $level); @@ -394,7 +414,8 @@ function is_error_message() { if (!function_exists('sql_save')) { function sql_save($array, $table, $key = 'id') { - return isset($array['id']) ? $array['id'] : 1; + $GLOBALS['__test_saved_rows'][] = ['table' => $table, 'row' => $array]; + return plugin_test_db_result('sql_save', isset($array['id']) ? $array['id'] : 1); } } @@ -471,3 +492,84 @@ function flowview_test_load($path) { require_once dirname(__DIR__) . '/includes/functions.php'; require_once dirname(__DIR__) . '/includes/database.php'; + +// Core helper doubles are reset through plugin_test_reset(), never test collection. +if (!function_exists('read_user_setting')) { + function read_user_setting($name, $default = false) { + return $GLOBALS['__test_user_settings'][$name] ?? $default; + } +} +if (!function_exists('get_timespan')) { + function get_timespan(&$span, $now, $timespan, $weekday) { + $result = plugin_test_db_result('get_timespan', ['begin_now' => $now - 1800, 'end_now' => $now]); + if ($result instanceof Throwable) { + throw $result; + } + $span = $result; + } +} +if (!function_exists('exec_with_timeout')) { + function exec_with_timeout($command, &$output, &$status, $timeout = 5) { + $GLOBALS['__test_db_calls'][] = ['fn' => 'exec_with_timeout', 'sql' => '', 'params' => [$command, $timeout]]; + $exit = plugin_test_db_result('exec_with_timeout', 0); + if ($exit !== null) { + $status = $exit; + } + return plugin_test_db_result('exec_with_timeout_result', null); + } +} + +if (!function_exists('db_affected_rows')) { + function db_affected_rows($conn = false) { + return plugin_test_db_result('db_affected_rows', 1); + } +} +if (!function_exists('register_process_start')) { + function register_process_start($type, $name, $id = 0, $timeout = 300) { + $GLOBALS['__test_db_calls'][] = ['fn' => 'register_process_start', 'sql' => '', 'params' => [$type, $name, $id, $timeout]]; + return plugin_test_db_result('register_process_start', true); + } +} + +if (!function_exists('reports_load_format_file')) { + function reports_load_format_file($format, &$output, &$report_tag, &$theme) { + $GLOBALS['__test_db_calls'][] = ['fn' => 'reports_load_format_file', 'sql' => '', 'params' => [$format]]; + $output = ''; + $report_tag = plugin_test_db_result('delivery_format_tag', true); + return plugin_test_db_result('delivery_format_ok', true); + } +} + +// Schema-mutation helpers are loaded explicitly for isolated upgrade tests. +require_once dirname(__DIR__) . '/includes/filter_upgrade.php'; + +if (!function_exists('cacti_session_close')) { + function cacti_session_close() { + $GLOBALS['__test_db_calls'][] = ['fn' => 'cacti_session_close', 'sql' => '', 'params' => []]; + } +} +if (!function_exists('array_rekey')) { + function array_rekey($rows, $key, $value) { + if (!is_array($value)) { + return array_column($rows, $value, $key); + } + $result = []; + foreach ($rows as $row) { + $result[$row[$key]] = array_intersect_key($row, array_flip($value)); + } + return $result; + } +} + +if (!function_exists('db_debug')) { + function db_debug($message) { + $GLOBALS['__test_db_calls'][] = ['fn' => 'db_debug', 'sql' => '', 'params' => [$message]]; + } +} + +if (!function_exists('mailer')) { + function mailer(...$args) { + $GLOBALS['__test_db_calls'][] = ['fn' => 'mailer', 'sql' => '', 'params' => $args]; + return plugin_test_db_result('mailer', ''); + } +} diff --git a/tests/fixtures/completion-budget.php b/tests/fixtures/completion-budget.php new file mode 100644 index 0000000..e6ed513 --- /dev/null +++ b/tests/fixtures/completion-budget.php @@ -0,0 +1,46 @@ + $argv[2]]; +$db = new PDO('sqlite::memory:'); +$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); +$db->exec('CREATE TABLE reports_queued (id INTEGER PRIMARY KEY, source TEXT, source_id INTEGER, status TEXT, start_time TEXT, scheduled_time TEXT, run_command TEXT, run_timeout INTEGER)'); +for ($id = 1; $id <= 101; $id++) { + $stmt = $db->prepare("INSERT INTO reports_queued VALUES (?, 'flowview', ?, ?, '', '', 'fixture', 1)"); + $stmt->execute([$id, $id, $id === 101 ? 'pending' : 'complete']); +} +$attempts = []; +$deletes = []; +function db_fetch_row(...$args) { return ['Field' => 'enabled']; } +function db_fetch_assoc_prepared($sql, $params = [], ...$args) { + if (str_contains($sql, 'FROM processes')) { return []; } + $stmt = $GLOBALS['db']->prepare($sql); + $stmt->execute($params); + return $stmt->fetchAll(PDO::FETCH_ASSOC); +} +function db_fetch_row_prepared($sql, $params = [], ...$args) { return db_fetch_assoc_prepared($sql, $params)[0] ?? []; } +function db_execute_prepared($sql, $params = [], ...$args) { + if (str_starts_with($sql, 'DELETE FROM reports_queued')) { + $GLOBALS['deletes'][] = $params[0]; + if ($params[0] <= 100) { return false; } + } + $stmt = $GLOBALS['db']->prepare($sql); + $result = $stmt->execute($params); + $GLOBALS['affected'] = $stmt->rowCount(); + return $result; +} +function db_affected_rows(...$args) { return $GLOBALS['affected'] ?? 0; } +function exec_with_timeout($command, &$output, &$status, $timeout) { + preg_match('/--queue-id=(\d+)/', $command, $match); + $id = (int) $match[1]; + $GLOBALS['attempts'][] = $id; + db_execute_prepared('UPDATE reports_queued SET status=? WHERE id=?', ['complete', $id]); + $status = 0; +} +function read_config_option(...$args) { return ''; } +function cacti_log(...$args) {} +require $argv[1] . '/includes/database.php'; +require $argv[1] . '/includes/functions.php'; +$done = flowview_reports_run_pending(1, 100); +$remaining = db_fetch_assoc_prepared('SELECT id,status FROM reports_queued ORDER BY id'); +echo json_encode(compact('done', 'attempts', 'deletes', 'remaining')); diff --git a/tests/fixtures/filter-lifecycle.php b/tests/fixtures/filter-lifecycle.php new file mode 100644 index 0000000..5a8351d --- /dev/null +++ b/tests/fixtures/filter-lifecycle.php @@ -0,0 +1,152 @@ + dirname(__DIR__), 'poller_id' => 1]; +$local_db_cnn_id = 'fixture'; +$mode = getenv('FLOWVIEW_FIXTURE_MODE'); +if ($mode === 'connection-failure') { + define('FLOWVIEW_CONFIG', dirname(__DIR__) . '/connection.php'); +} +$ready = in_array($mode, ['ready', 'late', 'child-disabled', 'report-read-failure', 'schedule-read-failure', 'filter-read-failure', 'parallel-timeout', 'timestamp-failure', 'master-busy', 'notification-read-failure', 'notification-missing', 'notification-after-delivery', 'notification-mail-error', 'notification-list-error', 'notification-log-error', 'notification-success'], true); +$metadataReads = 0; +$calls = []; +const MESSAGE_LEVEL_INFO = 1; +const MESSAGE_LEVEL_ERROR = 2; +function __(string $text, ...$args) { + return $text; +} +function html_escape($text) { + return htmlspecialchars($text, ENT_QUOTES, 'UTF-8'); +} +function cacti_sizeof($value) { + return is_array($value) ? count($value) : 0; +} +function read_config_option($name) { + if (str_starts_with($GLOBALS['mode'], 'notification-') && $name === 'flowview_parallel_threads') { return 1; } + return ['path_php_binary' => PHP_BINARY, 'reports_concurrent' => 1, 'flowview_parallel_threads' => 2, 'flowview_partition' => 1, 'flowview_parallel_time_to_live' => 21600, 'flowview_parallel_runlimit' => 0][$name] ?? ''; +} +function cacti_log(...$args) { +} +function raise_message(...$args) { +} +function db_fetch_cell_prepared($sql, ...$args) { + if (str_contains($sql, 'SELECT name')) { + return 'Fixture'; + } + if (str_contains($sql, 'SELECT id')) { + return 71; + } + return str_contains($sql, 'parallel_database_query') ? '' : 'fixture-version'; +} +function db_fetch_cell(...$args) { + return str_contains($args[0], 'SELECT version') ? 'fixture-version' : 1; +} +function db_fetch_row($sql, ...$args) { + $GLOBALS['calls'][] = ['read', $sql]; + if (str_contains($sql, 'MIN(start_time)')) { + return ['min_date' => '2024-01-01 00:00:00', 'max_date' => '2024-01-01 01:00:00', 'table_partition' => '']; + } + $GLOBALS['metadataReads']++; + return $GLOBALS['ready'] && ($GLOBALS['mode'] !== 'late' || $GLOBALS['metadataReads'] === 1) ? ['Field' => 'enabled'] : []; +} +function db_execute($sql, ...$args) { + $GLOBALS['calls'][] = ['ddl', $sql]; + $GLOBALS['ready'] = $GLOBALS['mode'] !== 'failed'; + return $GLOBALS['ready']; +} +function exec_background(...$args) { + $GLOBALS['calls'][] = ['background', $args]; +} +function register_process_start(...$args) { + $GLOBALS['calls'][] = ['register', $args]; + return !in_array($GLOBALS['mode'], ['busy', 'master-busy'], true); +} +function unregister_process(...$args) { + $GLOBALS['calls'][] = ['unregister', $args]; +} +function set_default_action() { + $GLOBALS['calls'][] = ['page-ready']; + exit; +} +// Capture state after shutdown callbacks registered by the actual entrypoint. +register_shutdown_function(function () { + register_shutdown_function(function () { + file_put_contents(getenv('FLOWVIEW_FIXTURE_RESULT'), json_encode(['calls' => $GLOBALS['calls'], 'status' => http_response_code(), 'repair_loaded' => function_exists('flowview_filter_status_upgrade'), 'pid' => getmypid(), 'native_helper_loaded' => getenv('FLOWVIEW_FIXTURE_NATIVE_HELPER') ? realpath((new ReflectionFunction('reports_log_and_notify'))->getFileName()) === realpath(getenv('FLOWVIEW_FIXTURE_NATIVE_HELPER')) : false])); + }); +}); +function db_fetch_row_prepared($sql, $params = [], ...$args) { + $GLOBALS['calls'][] = ['row', $sql, $params]; + if (($GLOBALS['mode'] === 'report-read-failure' && str_contains($sql, 'reports_queued')) + || ($GLOBALS['mode'] === 'schedule-read-failure' && str_contains($sql, 'plugin_flowview_schedules')) + || ($GLOBALS['mode'] === 'filter-read-failure' && str_contains($sql, 'plugin_flowview_queries'))) { + return false; + } + if (str_contains($sql, 'reports_queued')) { + $reads = array_filter($GLOBALS['calls'], fn ($c) => $c[0] === 'row' && str_contains($c[1], 'reports_queued')); + if (count($reads) > 1 && $GLOBALS['mode'] === 'notification-read-failure') { return false; } + if (count($reads) > 1 && $GLOBALS['mode'] === 'notification-missing') { return []; } + if (in_array($GLOBALS['mode'], ['notification-after-delivery', 'notification-mail-error', 'notification-log-error', 'notification-success', 'notification-list-error'], true)) { + return ['id' => 51, 'source' => 'flowview', 'source_id' => 7, 'name' => 'Fixture', 'notification' => ($GLOBALS['mode'] === 'notification-list-error' || getenv('FLOWVIEW_FIXTURE_RECIPIENTS')) ? '{"notification_list":{"id":1}}' : '{"email":{"to_email":"fixture@example.invalid"}}', 'request_type' => 'scheduled', 'requested_by' => 'fixture', 'requested_id' => 1]; + } + return ['id' => 51, 'source' => 'flowview', 'source_id' => 7]; + } + if (str_contains($sql, 'plugin_notification_lists')) { + return getenv('FLOWVIEW_FIXTURE_RECIPIENTS') ? json_decode(getenv('FLOWVIEW_FIXTURE_RECIPIENTS'), true) : ['emails' => 'list@example.invalid', 'format_file' => 'default']; + } + if (str_contains($sql, 'plugin_flowview_schedules')) { + return ['id' => 7, 'query_id' => 3, 'title' => 'Fixture', 'format_file' => 'default']; + } + return ($GLOBALS['mode'] === 'parallel-timeout' || str_starts_with($GLOBALS['mode'], 'notification-')) + ? ['id' => 3, 'name' => 'Fixture', 'enabled' => 'on', 'timespan' => 1, 'statistics' => 9, 'printed' => 0, 'sortfield' => 'bytes', 'cutofflines' => 100] + : ['enabled' => '']; +} +function db_execute_prepared($sql, $params = [], ...$args) { + $GLOBALS['calls'][] = ['write', $sql, $params]; + return !($GLOBALS['mode'] === 'timestamp-failure' && str_contains($sql, 'UPDATE plugin_flowview_schedules')); +} + +function db_fetch_assoc($sql, ...$args) { + return str_contains($sql, 'information_schema') ? [['table' => 'plugin_flowview_raw_202400000']] : []; +} +function db_fetch_assoc_prepared(...$args) { + return str_starts_with($GLOBALS['mode'], 'notification-') ? [['bytes' => 42]] : []; +} +function get_timespan(&$span, ...$args) { + $span = ['begin_now' => 1704067200, 'end_now' => 1704070800]; + if (str_starts_with($GLOBALS['mode'], 'notification-')) { + $span = ['begin_now' => 0, 'end_now' => 2000]; + $key = get_flowview_session_key(3, 100, 2000); + $_SESSION['sess_flowdata'][$key] = ['timeout' => time() + 60, 'data' => ['table' => 'Fixture', 'data' => [['bytes' => 42]]]]; + } +} +function read_user_setting(...$args) { return ''; } +function isset_request_var(...$args) { return str_starts_with($GLOBALS['mode'], 'notification-') && in_array($args[0], ['timespan', 'query', 'date1', 'date2'], true); } +function get_request_var(...$args) { return ['timespan' => 'custom', 'query' => 3, 'date1' => date('Y-m-d H:i:s', 100), 'date2' => date('Y-m-d H:i:s', 2000)][$args[0]] ?? null; } +function get_nfilter_request_var(...$args) { return null; } +function get_filter_request_var(...$args) { return null; } +function cacti_session_close() {} +function raise_message_javascript(...$args) {} +function array_rekey($rows, $key, $value) { + return array_column($rows, $value, $key); +} + +function __esc($text, ...$args) { return html_escape($text); } + +function db_connect_real(...$args) { + $GLOBALS['calls'][] = ['connect']; + return false; +} + +function reports_load_format_file($format, &$output, &$tag, &$theme) { $output = ''; $tag = true; return true; } +function mailer(...$args) { + $GLOBALS['calls'][] = ['delivery', array_slice($args, 1, 3)]; + if ($GLOBALS['mode'] === 'notification-after-delivery') { throw new RuntimeException('Ambiguous delivery failure'); } + return in_array($GLOBALS['mode'], ['notification-mail-error', 'notification-list-error'], true) ? 'SMTP delivery failed' : ''; +} +function sql_save(...$args) { $GLOBALS['calls'][] = ['save-log']; return $GLOBALS['mode'] === 'notification-log-error' ? false : 1; } + +if (getenv('FLOWVIEW_FIXTURE_NATIVE_HELPER')) { + define('CACTI_VERSION', '1.3.0'); + function cacti_version_compare($a, $b, $op) { return version_compare($a, $b, $op); } + require getenv('FLOWVIEW_FIXTURE_NATIVE_HELPER'); +} diff --git a/tests/fixtures/filter-queue.php b/tests/fixtures/filter-queue.php new file mode 100644 index 0000000..7ea7da9 --- /dev/null +++ b/tests/fixtures/filter-queue.php @@ -0,0 +1,70 @@ + $argv[2]]; +$db = new PDO('sqlite::memory:'); +$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); +$db->exec('CREATE TABLE reports_queued (id INTEGER PRIMARY KEY, source TEXT, source_id INTEGER, status TEXT, start_time TEXT, scheduled_time TEXT, run_command TEXT, run_timeout INTEGER)'); +for ($id = 1; $id <= 4; $id++) { + $db->exec("INSERT INTO reports_queued VALUES ($id, 'flowview', $id, 'pending', '0000-00-00 00:00:00', '2026-01-01 00:00:00', 'fixture', 1)"); +} +$attempts = []; +function db_fetch_row(...$args) { + return ['Field' => 'enabled']; +} +function db_fetch_assoc_prepared($sql, $params = [], ...$args) { + if (str_contains($sql, 'FROM processes')) { + return []; + } + $stmt = $GLOBALS['db']->prepare($sql); + $stmt->execute($params); + return $stmt->fetchAll(PDO::FETCH_ASSOC); +} +function db_fetch_row_prepared($sql, $params = [], ...$args) { + return db_fetch_assoc_prepared($sql, $params)[0] ?? []; +} +function db_execute_prepared($sql, $params = [], ...$args) { + $stmt = $GLOBALS['db']->prepare($sql); + $result = $stmt->execute($params); + $GLOBALS['affected'] = $stmt->rowCount(); + return $result; +} +function db_affected_rows(...$args) { + return $GLOBALS['affected'] ?? 0; +} +function exec_with_timeout($command, &$output, &$status, $timeout) { + preg_match('/--queue-id=(\d+)/', $command, $match); + $id = (int) $match[1]; + $GLOBALS['attempts'][] = $id; + $status = $id <= 2 || $id === 6 ? 75 : 0; + if ($status === 0) { + db_execute_prepared('UPDATE reports_queued SET status=? WHERE id=?', ['complete', $id]); + } + if ($status === 75) { + db_execute_prepared('UPDATE reports_queued SET status=? WHERE id=?', ['retry', $id]); + if ($id === 6) { + // Another dispatcher consumed the child's retry marker before this parent returned. + db_execute_prepared('UPDATE reports_queued SET status=? WHERE id=?', ['running', $id]); + } + } + return ''; +} +function read_config_option(...$args) { + return ''; +} +function cacti_log(...$args) { +} +require $argv[1] . '/includes/database.php'; +require $argv[1] . '/includes/functions.php'; +$first = flowview_reports_run_pending(1, 2); +$afterFirst = db_fetch_assoc_prepared('SELECT id,status FROM reports_queued ORDER BY id'); +$second = flowview_reports_run_pending(1, 2); +$remaining = db_fetch_assoc_prepared('SELECT id,status FROM reports_queued ORDER BY id'); +$third = flowview_reports_run_pending(1, 2); +$db->exec("INSERT INTO reports_queued VALUES (5, 'flowview', 5, 'running', '2026-01-01 00:00:00', '2026-01-01 00:00:00', 'fixture', 1)"); +$lostClaim = flowview_reports_run(5); +$claimedStatus = db_fetch_row_prepared('SELECT status FROM reports_queued WHERE id=5'); +$db->exec("INSERT INTO reports_queued VALUES (6, 'flowview', 6, 'pending', '2026-01-01 00:00:00', '2026-01-01 00:00:00', 'fixture', 1)"); +$retryRace = flowview_reports_run(6); +$raceStatus = db_fetch_row_prepared('SELECT status FROM reports_queued WHERE id=6'); +echo json_encode(compact('retryRace', 'raceStatus', 'first', 'second', 'third', 'attempts', 'afterFirst', 'remaining', 'lostClaim', 'claimedStatus')); diff --git a/tests/fixtures/parallel-retry.php b/tests/fixtures/parallel-retry.php new file mode 100644 index 0000000..3930d06 --- /dev/null +++ b/tests/fixtures/parallel-retry.php @@ -0,0 +1,85 @@ + $argv[2]]; +$debug = false; +$db = new PDO('sqlite::memory:'); +$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); +$db->exec("CREATE TABLE parallel_database_query (id INTEGER PRIMARY KEY AUTOINCREMENT, md5sum TEXT, md5sum_tables TEXT, user_id INTEGER, total_shards INTEGER, map_query TEXT, map_range TEXT, map_range_params TEXT, reduce_query TEXT, created TEXT, time_to_live INTEGER, map_table TEXT, map_create TEXT, status TEXT DEFAULT 'pending', results TEXT)"); +$db->exec('CREATE TABLE parallel_database_query_shard (query_id INTEGER, shard_id INTEGER, full_scan INTEGER, map_table TEXT, map_partition TEXT, map_query TEXT, map_params TEXT)'); +$phase = 0; +$launches = []; +const MESSAGE_LEVEL_ERROR = 1; +const POLLER_VERBOSITY_MEDIUM = 2; +function __($text, ...$args) { return $text; } +function cacti_sizeof($value) { return is_array($value) ? count($value) : 0; } +function cacti_log(...$args) {} +function raise_message_javascript(...$args) {} +function read_config_option($name) { + return ['path_php_binary' => PHP_BINARY, 'flowview_parallel_runlimit' => $GLOBALS['phase'] ? 5 : 0, 'flowview_parallel_time_to_live' => 21600, 'flowview_engine' => 'Aria'][$name] ?? ''; +} +function array_rekey($rows, $key, $value) { + $result = []; + foreach ($rows as $row) { + $result[$row[$key]] = is_array($value) ? array_intersect_key($row, array_flip($value)) : $row[$value]; + } + return $result; +} +function db_fetch_assoc_prepared($sql, $params = [], ...$args) { + if (str_contains($sql, 'FROM processes')) { + return []; + } + if (str_contains($sql, 'INFORMATION_SCHEMA.COLUMNS')) { + return [['COLUMN_NAME' => 'bytes', 'DATA_TYPE' => 'bigint', 'COLUMN_TYPE' => 'bigint', 'IS_NULLABLE' => 'YES', 'COLUMN_DEFAULT' => '']]; + } + $stmt = $GLOBALS['db']->prepare($sql); + $stmt->execute($params); + return $stmt->fetchAll(PDO::FETCH_ASSOC); +} +function db_fetch_assoc($sql, ...$args) { + return str_contains($sql, 'information_schema.ENGINES') ? [['ENGINE' => 'Aria']] : db_fetch_assoc_prepared($sql); +} +function db_fetch_cell_prepared($sql, $params = [], ...$args) { + $rows = db_fetch_assoc_prepared($sql, $params); + return $rows ? reset($rows[0]) : false; +} +function db_fetch_cell($sql, ...$args) { return db_fetch_cell_prepared($sql); } +function db_execute_prepared($sql, $params = [], ...$args) { + $stmt = $GLOBALS['db']->prepare($sql); + return $stmt->execute($params); +} +function db_execute($sql, ...$args) { + $sql = preg_replace('/\) ENGINE=.*$/s', ')', $sql); + return db_execute_prepared($sql); +} +function sql_save($row, $table, ...$args) { + unset($row['id']); + $sql = 'INSERT INTO ' . $table . ' (' . implode(',', array_keys($row)) . ') VALUES (' . implode(',', array_fill(0, count($row), '?')) . ')'; + db_execute_prepared($sql, array_values($row)); + return (int) $GLOBALS['db']->lastInsertId(); +} +function exec_background($php, $command, ...$args) { + preg_match('/--query-id=(\d+)/', $command, $match); + $id = (int) $match[1]; + $GLOBALS['launches'][] = $id; + if ($GLOBALS['phase']) { + db_execute_prepared('UPDATE parallel_database_query SET status=?,results=? WHERE id=?', ['complete', '[]', $id]); + } +} +require $argv[1] . '/includes/database.php'; +require $argv[1] . '/includes/functions.php'; +$tables = ['fixture_raw' => ['min_date' => '2024-01-01 00:00:00', 'max_date' => '2024-01-01 01:00:00', 'table_name' => 'fixture_raw', 'table_partition' => '']]; +$inner = ['sql_query' => 'SELECT SUM(bytes) bytes', 'sql_where' => '', 'sql_range' => 'start_time >= ? AND start_time < ?', 'sql_range_params' => [1704067200, 1704070800], 'sql_params' => [], 'sql_start_time' => 1704067200, 'sql_end_time' => 1704070800]; +$outer = ['sql_query' => 'SELECT SUM(bytes) bytes', 'sql_where' => '', 'sql_groupby' => '', 'sql_having' => '', 'sql_order' => '', 'sql_limit' => '', 'sql_params' => []]; +$first = parallel_database_query_request($tables, $inner, $outer); +$timeout = parallel_database_query_run([$first]); +$remaining = db_fetch_cell('SELECT COUNT(*) FROM parallel_database_query'); +$remainingShards = db_fetch_cell('SELECT COUNT(*) FROM parallel_database_query_shard'); +$phase = 1; +$retry = parallel_database_query_request($tables, $inner, $outer); +$retryShards = db_fetch_cell_prepared('SELECT COUNT(*) FROM parallel_database_query_shard WHERE query_id=?', [$retry]); +$result = parallel_database_query_run([$retry]); +$cached = parallel_database_query_request($tables, $inner, $outer); +$cachedResult = parallel_database_query_run([$cached]); +echo json_encode(compact('first', 'timeout', 'remaining', 'remainingShards', 'retry', 'retryShards', 'result', 'cached', 'cachedResult', 'launches')); diff --git a/tests/fixtures/report-stderr.php b/tests/fixtures/report-stderr.php new file mode 100644 index 0000000..2239451 --- /dev/null +++ b/tests/fixtures/report-stderr.php @@ -0,0 +1,15 @@ + 'enabled']); +plugin_test_queue_db_result('db_fetch_row_prepared', ['id' => 51, 'source' => 'flowview', 'name' => 'Fixture', 'source_id' => 7, 'run_command' => $command, 'run_timeout' => 5]); +plugin_test_queue_db_result('report_outcome', ['status' => 'complete']); +$success = $argv[4] === 'native' ? flowview_reports_run(51) : reports_run(51); +$writes = array_values(array_filter($GLOBALS['__test_db_calls'], fn ($c) => $c['fn'] === 'db_execute_prepared')); +echo json_encode(compact('helperResult', 'status', 'success', 'writes')); diff --git a/tests/fixtures/report-timeout.php b/tests/fixtures/report-timeout.php new file mode 100644 index 0000000..7243118 --- /dev/null +++ b/tests/fixtures/report-timeout.php @@ -0,0 +1,47 @@ + $argv[1]]; +$dbPath = $argv[3] . '/queue.sqlite'; +$db = new PDO('sqlite:' . $dbPath); +$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); +$db->exec('CREATE TABLE reports_queued (id INTEGER PRIMARY KEY, source TEXT, source_id INTEGER, status TEXT, start_time TEXT, scheduled_time TEXT, run_command TEXT, run_timeout INTEGER)'); +$worker = $argv[3] . '/late.php'; +$delivery = $argv[3] . '/deliveries.txt'; +file_put_contents($worker, 'exec("UPDATE reports_queued SET status=\'complete\' WHERE id=51 AND status=\'running\'");'); +$command = escapeshellarg(PHP_BINARY) . ' ' . escapeshellarg($worker); +$stmt = $db->prepare("INSERT INTO reports_queued VALUES (51,'flowview',7,'pending','','',?,1)"); +$stmt->execute([$command]); +$writes = []; +$logs = []; +function db_fetch_row($sql, ...$args) { return ['Field' => 'enabled']; } +function db_fetch_assoc_prepared($sql, $params = [], ...$args) { + if (str_contains($sql, 'FROM processes')) { return []; } + $stmt = $GLOBALS['db']->prepare($sql); + $stmt->execute($params); + return $stmt->fetchAll(PDO::FETCH_ASSOC); +} +function db_fetch_row_prepared($sql, $params = [], ...$args) { return db_fetch_assoc_prepared($sql, $params)[0] ?? []; } +function db_execute_prepared($sql, $params = [], ...$args) { + $GLOBALS['writes'][] = [$sql, $params]; + $stmt = $GLOBALS['db']->prepare($sql); + $result = $stmt->execute($params); + $GLOBALS['affected'] = $stmt->rowCount(); + return $result; +} +function db_affected_rows(...$args) { return $GLOBALS['affected'] ?? 0; } +function cacti_sizeof($rows) { return is_array($rows) ? count($rows) : 0; } +function read_config_option(...$args) { return ''; } +function cacti_log($text, ...$args) { $GLOBALS['logs'][] = $text; } +require $argv[1] . '/lib/poller.php'; +require $argv[2] . '/includes/database.php'; +require $argv[2] . '/includes/functions.php'; +$initial = $argv[4] === 'native' ? flowview_reports_run(51) : reports_run(51); +$deadline = microtime(true) + 5; +while (!is_file($delivery) && microtime(true) < $deadline) { usleep(50000); } +// The delivery marker precedes the durable outcome; wait for that write too. +while ((db_fetch_row_prepared('SELECT status FROM reports_queued WHERE id=51')['status'] ?? '') === 'running' && microtime(true) < $deadline) { usleep(50000); } +$late = flowview_reports_run_pending(); +$remaining = db_fetch_assoc_prepared('SELECT * FROM reports_queued'); +$deliveries = is_file($delivery) ? file($delivery) : []; +echo json_encode(compact('initial', 'late', 'remaining', 'deliveries', 'writes', 'logs'));