Automation architect

Monitor the stack queue

The stack runs the asynchronous scripts. The script reads its limit of threads, the waiting, running, closed and failed executions, the threads in use, the error rate and the stack door, and returns ALERT when the stack is saturated or when more than 5 % of the runs fail.

#1. Counters of the stack;
-> "[limit]" (stack process_limit);
-> "[wait]" (stack count_wait);
-> "[run]" (stack count_running);
-> "[closed]" (stack count_closed);
-> "[err]" (stack count_error);
#2. Threads in use, read in the status text;
-> "[inuse]" (string regexp_extract (stack status) "NB_PROCESS_IN_STACK=([0-9]+)" 1);
#3. Error rate, saturation and the stack door;
-> "[rate]" (if (> (+ [closed] [err]) 0) {math round (* 100 (/ [err] (+ [closed] [err]))) 2} {0});
json load "rx_doors" (string substring (maintenance get) 39);
-> "[verdict]" (if (or (>= [inuse] [limit]) (> [rate] 5)) {"ALERT"} {"OK"});
concat "{\"limit\": " [limit] ", \"in_use\": " [inuse] ", \"wait\": " [wait] ", \"running\": " [run] ", \"closed\": " [closed] ", \"error\": " [err] ", \"error_rate_pct\": " [rate] ", \"stack_locked\": " (json select "rx_doors" /stack) ", \"scheduler\": \"" (job scheduler status) "\", \"verdict\": \"" [verdict] "\"}"

{"limit": 10, "in_use": 0, "wait": 0, "running": 0, "closed": 0, "error": 0, "error_rate_pct": 0, "stack_locked": false, "scheduler": "Running...", "verdict": "OK"}

Audit of the cron jobs

Which jobs are scheduled, which ones are paused, how many fire every 30 seconds, when the backup runs, and the exact request to change the metric job. Session variables count the active and inactive jobs.

#1. All the jobs, and the active ones;
json load "rx_jobs" (job show);
json load "rx_act" (job show activate);
json load "rx_ids" (json fields "rx_jobs" /);
#2. Session counters: active jobs, inactive jobs, paused list;
env set var "[active]" 0;
env set var "[inactive]" (json count "rx_ids" /);
for (-> "[i]" 0) (< [i] (json count "rx_ids" /)) (++ "[i]") {
	-> "[id]" (json select "rx_ids" (concat "/[" [i] "]"));
	if (equal (json select "rx_jobs" (concat "/" [id] "/activate")) "1") {
		env incr var "[active]" 1;
		env decr var "[inactive]" 1;
	} {concat_var "[paused]" [id] " "};
};
#3. Frequent jobs, the backup schedule and the update request of the metric job;
-> "[every30s]" (string regexp_count (job show) "0/30 \* \* \* \* \?");
-> "[report]" (concat (json count "rx_ids" /) " jobs, " (env get var "[active]") " active, " (json count "rx_act" /) " scheduled, " (env get var "[inactive]") " inactive: " (if (env exist var "[paused]") {[paused]} {"-"}) [_n_] "every 30 s: " [every30s] ", backup: " (json select "rx_jobs" "/server.backup/pattern") ", metric job exists: " (job exist "server.metric") [_n_] (job generate update "server.metric"));
env del var "[paused]";
[report]

14 jobs, 11 active, 11 scheduled, 3 inactive: users_mail_auto innov_ai_supervision_2 innov_ai_supervision_1
every 30 s: 3, backup: 0 0 4 * * ?, metric job exists: 1
job update "server.metric" "server.metric.post" "0 * * * * ?" true;
job pause "server.metric";
job resume "server.metric";

From a script to its stack call

Before wiring a worker in a trigger, the architect checks that the script exists, reads its parameters, runs it in the three ways (new environment, parent environment, inline) and generates the asynchronous call for the stack.

#1. The worker script exists and its parameters;
-> "[s]" "addition.post";
-> "[exists]" (script exist [s]);
json load "rx_p" (script show params [s]);
json load "rx_def" (script get [s]);
json load "rx_ghost" (script show ghost);
-> "[names]" (concat (json select "rx_p" "/[0]/name") " " (json select "rx_p" "/[1]/name"));
#2. Run it now: new environment, parent environment, inline;
-> "[r1]" (execute [s] "[v1]" 12 "[v2]" 30);
-> "[r2]" (call [s] "[v1]" 1.5 "[v2]" 2.5);
-> "[r3]" (include [s] "[v1]" 100 "[v2]" -1);
json unload "rx_p";
-> "[cfg]" (concat "attempts=" (json select "rx_def" /nbAttempt) " log=" (json select "rx_def" /activateLog) " ghosts=" (json count "rx_ghost" /));
#3. Generate the asynchronous call to paste in a job or a trigger;
concat "exists=" [exists] ", params=" [names] ", execute=" [r1] ", call=" [r2] ", include=" [r3] ", " [cfg] [_n_] (script generate stack [s])

exists=1, params=[v1] [v2], execute=42, call=4, include=99, attempts=1 log=0 ghosts=0
stack_execute "2026-11-01 16:27:23" "addition.post"
	"[v1]" "example ..."
	"[v2]" "example ..."
;

Nightly import with a cron, a file watcher and a flow

Order files are dropped in a folder. A worker script tracks its stages as a stack flow, runs with priority 5 and 2 runs in parallel, can be replayed, and is started every night at 02:30 on one node of the cluster and on each new file.

#1. The drop folder is a parameter, not a constant in the code;
parameter merge "ETL_ORDERS_DIR" "/data/in/orders";
parameter set_description "ETL_ORDERS_DIR" "Drop folder of the daily order files";
#2. The worker tracks its stages as a stack flow;
script create exe "etl.orders.import" false 3
  (param
  	(var "[file]" {true} "The CSV file to import" is_null:true is_empty:true "orders_20261101.csv")
  )
  "Import the order files of the drop folder"
{
	stack flow_init [PID] "orders_import" "{}";
	stack flow_step [PID] 1 "parse";
	json load "flow" (stack flow_json_get [PID]);
	json iobject "flow" / "file" [file] STR;
	stack flow_json_set [PID] (json doc "flow");
	stack flow_step [PID] 2 "loaded";
	[file]
} "The imported file";
#3. Priority 5, 2 runs in parallel, replayable; retry window of 1 hour;
script set stack "etl.orders.import.exe" 5 2 true;
script set delay "etl.orders.import.exe" 1 hour {true};
#4. Every night at 02:30 on one node of the cluster, and on each new file;
job add "etl_orders_nightly" "etl.orders.import.exe" "0 30 2 * * ?" true;
job set cluster_script "etl_orders_nightly" "cluster.1n.job_execution.exe";
file_watcher start "orders_watch" "admin" (parameter get value "ETL_ORDERS_DIR") "etl.orders.import.exe";
#5. Check what is in place;
concat (job exist "etl_orders_nightly") " " (file_watcher exist "orders_watch") " " (parameter generate merge "ETL_ORDERS_DIR") (job generate update "etl_orders_nightly")

Retry the failed runs of a worker

Some order files failed. The architect queues two new files, finds the failed runs of the day, opens the newest one with its variables, pauses the cron, resets the attempts and replays the failed runs, drops the corrupt one and purges the closed runs.

#1. Queue two files for the worker, now;
stack (date now) "etl.orders.import.exe" "[file]" "orders_A.csv";
stack (date now) "etl.orders.import.exe" "[file]" "orders_B.csv";
#2. Executions in error since yesterday, newest first;
json load "errs" (string substring (stack search "error" "etl.orders.import.exe" "error" (date datedifft (concat (date sysdate) " 00:00:00") "DAY" 1) (concat (date sysdate) " 23:59:59") DESC 1 20) 39);
-> "[pid]" (json select "errs" "/data/[0]/pid");
stack get [pid];
stack var_show [pid];
#3. Pause the cron, give one more chance to all the failed runs and replay them;
job pause "etl_orders_nightly";
stack reset_error_nbattempt_script "etl.orders.import.exe";
stack replay_error_script "etl.orders.import.exe";
#4. The newest one is a corrupt file: reset it alone or drop it;
stack reset_error_nbattempt_id [pid];
stack replay_error_id [pid];
stack delete_error_id [pid];
stack delete_wait_script "etl.orders.import_old.exe";
stack process;
job resume "etl_orders_nightly";
#5. Purge the closed runs and give the counters;
stack delete_closed_script "etl.orders.import.exe";
concat "wait=" (stack count_wait) " running=" (stack count_running) " error=" (stack count_error)

Kill a runaway process of the stack

A long export runs in the stack and must be stopped (wrong parameters, window closed). The script pushes a 60-chunk export, waits until it really runs, kills its process and proves it stopped: the chunk counter does not move and the stack entry is in error with the cancel message.

#1. A long export that writes one chunk per second, 60 chunks;
sequence add "tuto_p4h_chunks" 0;
script create exe "tuto.p4h.export" false 1
	(param
		(var "[chunks]" {type is_number [chunks]} "Number of chunks" is_null:false is_empty:false "60")
	;)
	"Slow export, one chunk per second"
{
	for (-> "[c]" 0) (< [c] [chunks]) (++ "[c]") {
		sequence increment "tuto_p4h_chunks";
		wait 1000;
	};
	"export done";
} "export done";
#2. Push it in the stack and wait until it really runs;
-> "[pid]" (stack (date now) "tuto.p4h.export.exe" "[chunks]" 60);
-> "[i]" 0;
while (and (< [i] 45) (< (sequence get current "tuto_p4h_chunks") 3)) {wait 1000; ++ "[i]";};
-> "[running]" (concat "running after ~" [i] " s, chunks written: " (sequence get current "tuto_p4h_chunks"));
#3. The export is a runaway: kill the process, it stops at its next MQL command;
-> "[kill]" (kill_process [pid]);
wait 3000;
-> "[c1]" (sequence get current "tuto_p4h_chunks");
wait 3000;
-> "[c2]" (sequence get current "tuto_p4h_chunks");
json load "rx_p4h_p" (stack get [pid]);
-> "[after]" (concat "chunks 3 s after the kill: " [c1] ", 6 s after: " [c2] " | stack: in error since " (string left (json select "rx_p4h_p" "/dterror") 19) ", attempt " (json select "rx_p4h_p" "/nbattempt") "/" (json select "rx_p4h_p" "/nbmaxattempt") ", message: " (string regexp_extract (json select "rx_p4h_p" "/lasterrormsg") "Sorry, ([^.]+)" 1));
#4. Clean up;
stack delete_error_id [pid];
stack delete_closed_id [pid];
script delete "tuto.p4h.export.exe";
sequence remove "tuto_p4h_chunks";
concat [running] [_n_] "kill_process: " [kill] [_n_] [after]

running after ~20 s, chunks written: 3
kill_process: The process will be stoped at the next MQL command ...
chunks 3 s after the kill: 3, 6 s after: 3 | stack: in error since 2026-11-01 18:31:03, attempt 1/1, message: the action has been canceled

Heartbeat job: schedule, pause, resume, reschedule

A heartbeat script runs every 2 seconds from a cron job. The script proves the life cycle of the job: it counts the beats, pauses the job during a deployment (the counter stops), resumes it at a slower rate and keeps the restore line before removing everything.

#1. A heartbeat script that counts its runs;
sequence add "tuto_p4h_beats" 0;
script create exe "tuto.p4h.beat" false 1 (param ;) "Heartbeat of the integration platform" {
	sequence increment "tuto_p4h_beats";
} "The number of beats";
#2. Schedule it every 2 seconds;
job add "tuto_p4h_heartbeat" "tuto.p4h.beat.exe" "0/2 * * * * ?" true;
wait 7000;
-> "[b1]" (sequence get current "tuto_p4h_beats");
#3. Pause it during a deployment: the counter must not move;
job pause "tuto_p4h_heartbeat";
wait 5000;
-> "[b2]" (sequence get current "tuto_p4h_beats");
#4. Resume, then slow it down to every 3 seconds;
job resume "tuto_p4h_heartbeat";
job update "tuto_p4h_heartbeat" "tuto.p4h.beat.exe" "0/3 * * * * ?" true;
wait 7000;
-> "[b3]" (sequence get current "tuto_p4h_beats");
-> "[restore]" (job generate update "tuto_p4h_heartbeat");
#5. Clean up;
job remove "tuto_p4h_heartbeat";
script delete "tuto.p4h.beat.exe";
sequence remove "tuto_p4h_beats";
concat "beats after 7 s: " [b1] " | after 5 s paused: " [b2] " | 7 s after resume at 3 s: " [b3] [_n_] "restore line: " [restore] [_n_] "job still there: " (job exist "tuto_p4h_heartbeat")

beats after 7 s: 4 | after 5 s paused: 4 | 7 s after resume at 3 s: 8
restore line: job update "tuto_p4h_heartbeat" "tuto.p4h.beat.exe" "0/3 * * * * ?" true;
job still there: 0

Replay a failed run once the partner is back

An order push fails because the partner API is down (HTTP 503). The worker is replayable in the stack: once the API is back, the script resets the attempts of this run only and replays it, then waits for the closing.

#1. A worker that calls a partner API, simulated by a gate parameter;
parameter add "tuto_p4h_partner_api" "down" "Tutorial: state of the partner API" false;
script create exe "tuto.p4h.push_order" false 1
	(param
		(var "[order]" {type is_number [order]} "Order number" is_null:false is_empty:false "5001")
	;)
	"Push an order to the partner"
{
	if (equal (parameter get value "tuto_p4h_partner_api") "down") {exception (1) ("partner API unavailable (HTTP 503)")};
	concat "order " [order] " accepted";
} "A confirmation";
script set stack "tuto.p4h.push_order.exe" 5 1 true;
#2. Replayable in the stack, priority 5. First run in the stack fails: the API is down;
-> "[pid]" (stack (date now) "tuto.p4h.push_order.exe" "[order]" 5001);
-> "[i]" 0;
while (and (< [i] 45) (is null (json load select "rx_p4h_st" (stack get [pid]) "/dterror"))) {wait 1000; ++ "[i]";};
-> "[first]" (concat "run 1 after ~" [i] " s: " (string regexp_extract (json load select "rx_p4h_st" (stack get [pid]) "/lasterrormsg") "partner API unavailable \(HTTP [0-9]+\)" 0));
#3. The partner is back: reset the attempts and replay this run only;
parameter update "tuto_p4h_partner_api" "up";
stack reset_error_nbattempt_id [pid];
stack replay_error_id [pid];
-> "[i]" 0;
while (and (< [i] 45) (is null (json load select "rx_p4h_st" (stack get [pid]) "/dtclosed"))) {wait 1000; ++ "[i]";};
-> "[second]" (concat "replay closed after ~" [i] " s, attempts " (json load select "rx_p4h_st" (stack get [pid]) "/nbattempt"));
#4. Clean up;
stack delete_closed_id [pid];
script delete "tuto.p4h.push_order.exe";
parameter remove "tuto_p4h_partner_api";
concat [first] [_n_] [second]

run 1 after ~3 s: partner API unavailable (HTTP 503)
replay closed after ~29 s, attempts 1

Supplier invoices from an IMAP mailbox to the stack

Supplier invoices arrive by e-mail. The script downloads the unread mails from known suppliers with Invoice in the subject, moves them to Processed or Rejected in the mailbox, then queues one OCR import per mail in the stack.

#1. Download the unread mails of the supplier invoices mailbox, only PDFs from known suppliers;
json load "rx_inbox" (mail download imap "tmp/invoices_in"
	20 "unread" "Processed|Rejected" false true
	null null
	(mql {string ends_with [imap_from] "@supplier-a.com"})
	(mql {string contains [imap_subject] "Invoice"})
	{cm get "cm_invoices_imap";}
);
#2. One stack entry per mail: the OCR worker runs in parallel, with retries;
-> "[dir]" (json select "rx_inbox" "/Directory");
-> "[n]" 0;
if (> (json select "rx_inbox" "/NbReceived") 0) {
	json load "rx_files" (file dir_list [dir]);
	json parse_array "rx_files" "/" "[mail]" {
		if (string ends_with [mail] ".json") {
			stack (date now) "invoice.ocr.import.exe" "[mail_file]" (concat [dir] "/" [mail]);
			++ "[n]";
		};
	};
};
#3. Summary for the run log;
concat [n] " invoices queued from " (json select "rx_inbox" "/NbReceived") " mails, " (json select "rx_inbox" "/Unread") " still unread"

Bounce monitoring of a newsletter

The bounces of a newsletter come back to a POP3 mailbox. The script downloads and deletes them, counts the hard bounces (SMTP 550) and pauses the newsletter job above a 5% hard bounce rate.

#1. Download the bounces of the newsletter mailbox and delete them on the server;
json load "rx_bounce" (mail download pop3 "tmp/bounces"
	50 true
	{cm get "cm_bounces_pop3";}
);
#2. Count the delivery failures by file;
-> "[dir]" (json select "rx_bounce" "/Directory");
json load "rx_files" (file dir_list [dir]);
-> "[hard]" 0;
json parse_array "rx_files" "/" "[f]" {
	if (string contains (file load (concat [dir] "/" [f]) "utf-8") "550") {++ "[hard]";};
};
#3. Above 5 percent of hard bounces, pause the newsletter job;
-> "[rate]" (math round (* 100 (/ [hard] (math max 1 (json select "rx_bounce" "/NbReceived")))) 1);
if (> [rate] 5) {job pause "newsletter.weekly"};
concat (json select "rx_bounce" "/NbReceived") " bounces, " [hard] " hard (550), rate " [rate] "%"