-module(lightspeed@async@backpressure).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/async/backpressure.gleam").
-export([boundary/4, default_boundary/0, new/1, valid/1, enqueue/2, start_available/2, state_label/1, succeed/2, fail/3, cancel/3, retry/2, state/2, runtime_boundary/1, in_flight_count/1, queued_count/1, mode_label/1, error_label/1, tasks/1, signature/1]).
-export_type([adapter_mode/0, boundary/0, task_state/0, runtime_error/0, task/0, runtime/0]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
?MODULEDOC(" Async runtime boundaries for cancellation and backpressure-aware adapters.\n").
-type adapter_mode() :: push_pull | push_only | pull_only.
-type boundary() :: {boundary, adapter_mode(), integer(), integer(), integer()}.
-type task_state() :: {queued, integer()} |
{in_flight, integer(), integer()} |
{succeeded, integer()} |
{failed, integer(), binary()} |
{cancelled, integer(), binary()}.
-type runtime_error() :: {duplicate_task, binary()} |
{queue_saturated, binary(), integer()} |
{unknown_task, binary()} |
{invalid_transition, binary(), binary(), binary()}.
-type task() :: {task, binary(), task_state()}.
-opaque runtime() :: {runtime,
boundary(),
integer(),
list(task()),
integer(),
integer()}.
-file("src/lightspeed/async/backpressure.gleam", 57).
?DOC(" Build a boundary profile.\n").
-spec boundary(adapter_mode(), integer(), integer(), integer()) -> boundary().
boundary(Mode, Max_in_flight, Max_queued, Cancellation_budget_ms) ->
{boundary, Mode, Max_in_flight, Max_queued, Cancellation_budget_ms}.
-file("src/lightspeed/async/backpressure.gleam", 72).
?DOC(" Default boundary profile for mixed push/pull adapters.\n").
-spec default_boundary() -> boundary().
default_boundary() ->
{boundary, push_pull, 3, 6, 250}.
-file("src/lightspeed/async/backpressure.gleam", 82).
?DOC(" Build a runtime.\n").
-spec new(boundary()) -> runtime().
new(Boundary) ->
{runtime, Boundary, 1, [], 0, 0}.
-file("src/lightspeed/async/backpressure.gleam", 93).
?DOC(" Validate boundary settings.\n").
-spec valid(runtime()) -> boolean().
valid(Runtime) ->
((erlang:element(3, erlang:element(2, Runtime)) > 0) andalso (erlang:element(
4,
erlang:element(2, Runtime)
)
>= 0))
andalso (erlang:element(5, erlang:element(2, Runtime)) > 0).
-file("src/lightspeed/async/backpressure.gleam", 427).
-spec has_task(list(task()), binary()) -> boolean().
has_task(Tasks_rev, Key) ->
case Tasks_rev of
[] ->
false;
[Task | Rest] ->
case erlang:element(2, Task) =:= Key of
true ->
true;
false ->
has_task(Rest, Key)
end
end.
-file("src/lightspeed/async/backpressure.gleam", 100).
?DOC(" Enqueue one task with backpressure checks.\n").
-spec enqueue(runtime(), binary()) -> {runtime(),
{ok, nil} | {error, runtime_error()}}.
enqueue(Runtime, Key) ->
case has_task(erlang:element(4, Runtime), Key) of
true ->
{Runtime, {error, {duplicate_task, Key}}};
false ->
case erlang:element(6, Runtime) >= erlang:element(
4,
erlang:element(2, Runtime)
) of
true ->
{Runtime,
{error,
{queue_saturated,
Key,
erlang:element(4, erlang:element(2, Runtime))}}};
false ->
Seq = erlang:element(3, Runtime),
Task = {task, Key, {queued, Seq}},
{{runtime,
erlang:element(2, Runtime),
Seq + 1,
[Task | erlang:element(4, Runtime)],
erlang:element(5, Runtime),
erlang:element(6, Runtime) + 1},
{ok, nil}}
end
end.
-file("src/lightspeed/async/backpressure.gleam", 343).
-spec append(list(task()), list(task())) -> list(task()).
append(Left, Right) ->
case Left of
[] ->
Right;
[Entry | Rest] ->
[Entry | append(Rest, Right)]
end.
-file("src/lightspeed/async/backpressure.gleam", 339).
-spec prepend(list(task()), task()) -> list(task()).
prepend(Values, Value) ->
[Value | Values].
-file("src/lightspeed/async/backpressure.gleam", 314).
-spec pop_next_queued(list(task()), integer(), list(task())) -> gleam@option:option({integer(),
binary(),
list(task())}).
pop_next_queued(Tasks_rev, Now_ms, Seen_rev) ->
case Tasks_rev of
[] ->
none;
[Task | Rest] ->
case erlang:element(3, Task) of
{queued, Seq} ->
{some,
{Seq,
erlang:element(2, Task),
begin
_pipe = lists:reverse(Seen_rev),
_pipe@1 = prepend(
_pipe,
{task,
erlang:element(2, Task),
{in_flight, Seq, Now_ms}}
),
append(_pipe@1, Rest)
end}};
_ ->
pop_next_queued(Rest, Now_ms, [Task | Seen_rev])
end
end.
-file("src/lightspeed/async/backpressure.gleam", 287).
-spec start_loop(runtime(), integer(), list(binary())) -> {runtime(),
list(binary())}.
start_loop(Runtime, Now_ms, Started_rev) ->
case erlang:element(5, Runtime) >= erlang:element(
3,
erlang:element(2, Runtime)
) of
true ->
{Runtime, lists:reverse(Started_rev)};
false ->
case pop_next_queued(erlang:element(4, Runtime), Now_ms, []) of
none ->
{Runtime, lists:reverse(Started_rev)};
{some, {Seq, Key, Tasks_rev}} ->
Next = {runtime,
erlang:element(2, Runtime),
erlang:element(3, Runtime),
Tasks_rev,
erlang:element(5, Runtime) + 1,
erlang:element(6, Runtime) - 1},
start_loop(
Next,
Now_ms,
[<<<<Key/binary, ":"/utf8>>/binary,
(erlang:integer_to_binary(Seq))/binary>> |
Started_rev]
)
end
end.
-file("src/lightspeed/async/backpressure.gleam", 133).
?DOC(" Start queued tasks up to in-flight limit.\n").
-spec start_available(runtime(), integer()) -> {runtime(), list(binary())}.
start_available(Runtime, Now_ms) ->
start_loop(Runtime, Now_ms, []).
-file("src/lightspeed/async/backpressure.gleam", 246).
?DOC(" Stable task-state label.\n").
-spec state_label(task_state()) -> binary().
state_label(State) ->
case State of
{queued, _} ->
<<"queued"/utf8>>;
{in_flight, _, _} ->
<<"in_flight"/utf8>>;
{succeeded, _} ->
<<"succeeded"/utf8>>;
{failed, _, _} ->
<<"failed"/utf8>>;
{cancelled, _, _} ->
<<"cancelled"/utf8>>
end.
-file("src/lightspeed/async/backpressure.gleam", 388).
-spec in_flight_count_of(task_state()) -> integer().
in_flight_count_of(State) ->
case State of
{in_flight, _, _} ->
1;
_ ->
0
end.
-file("src/lightspeed/async/backpressure.gleam", 377).
-spec in_flight_delta(task_state(), task_state()) -> integer().
in_flight_delta(From, To) ->
in_flight_count_of(To) - in_flight_count_of(From).
-file("src/lightspeed/async/backpressure.gleam", 381).
-spec queued_count_of(task_state()) -> integer().
queued_count_of(State) ->
case State of
{queued, _} ->
1;
_ ->
0
end.
-file("src/lightspeed/async/backpressure.gleam", 373).
-spec queued_delta(task_state(), task_state()) -> integer().
queued_delta(From, To) ->
queued_count_of(To) - queued_count_of(From).
-file("src/lightspeed/async/backpressure.gleam", 420).
-spec reverse_into(list(task()), list(task())) -> list(task()).
reverse_into(Left, Right) ->
case Left of
[] ->
Right;
[Entry | Rest] ->
reverse_into(Rest, [Entry | Right])
end.
-file("src/lightspeed/async/backpressure.gleam", 395).
-spec update_task(
list(task()),
binary(),
fun((task_state()) -> {ok, task_state()} | {error, runtime_error()}),
list(task())
) -> {ok, {list(task()), task_state(), task_state()}} | {error, runtime_error()}.
update_task(Tasks_rev, Key, With, Seen_rev) ->
case Tasks_rev of
[] ->
{error, {unknown_task, Key}};
[Task | Rest] ->
case erlang:element(2, Task) =:= Key of
false ->
update_task(Rest, Key, With, [Task | Seen_rev]);
true ->
case With(erlang:element(3, Task)) of
{error, Error} ->
{error, Error};
{ok, Next_state} ->
{ok,
{reverse_into(
Seen_rev,
[{task,
erlang:element(2, Task),
Next_state} |
Rest]
),
erlang:element(3, Task),
Next_state}}
end
end
end.
-file("src/lightspeed/async/backpressure.gleam", 350).
-spec transition(
runtime(),
binary(),
binary(),
fun((task_state()) -> {ok, task_state()} | {error, runtime_error()})
) -> {runtime(), {ok, nil} | {error, runtime_error()}}.
transition(Runtime, Key, _, With) ->
case update_task(erlang:element(4, Runtime), Key, With, []) of
{error, Error} ->
{Runtime, {error, Error}};
{ok, {Tasks_rev, From_state, To_state}} ->
Runtime@1 = {runtime,
erlang:element(2, Runtime),
erlang:element(3, Runtime),
Tasks_rev,
erlang:element(5, Runtime) + in_flight_delta(
From_state,
To_state
),
erlang:element(6, Runtime) + queued_delta(From_state, To_state)},
{Runtime@1, {ok, nil}}
end.
-file("src/lightspeed/async/backpressure.gleam", 141).
?DOC(" Mark one in-flight task as succeeded.\n").
-spec succeed(runtime(), binary()) -> {runtime(),
{ok, nil} | {error, runtime_error()}}.
succeed(Runtime, Key) ->
transition(Runtime, Key, <<"succeed"/utf8>>, fun(State) -> case State of
{in_flight, Seq, _} ->
{ok, {succeeded, Seq}};
_ ->
{error,
{invalid_transition,
Key,
state_label(State),
<<"succeed"/utf8>>}}
end end).
-file("src/lightspeed/async/backpressure.gleam", 159).
?DOC(" Mark one in-flight task as failed.\n").
-spec fail(runtime(), binary(), binary()) -> {runtime(),
{ok, nil} | {error, runtime_error()}}.
fail(Runtime, Key, Reason) ->
transition(Runtime, Key, <<"fail"/utf8>>, fun(State) -> case State of
{in_flight, Seq, _} ->
{ok, {failed, Seq, Reason}};
_ ->
{error,
{invalid_transition,
Key,
state_label(State),
<<"fail"/utf8>>}}
end end).
-file("src/lightspeed/async/backpressure.gleam", 178).
?DOC(" Cancel one queued or in-flight task.\n").
-spec cancel(runtime(), binary(), binary()) -> {runtime(),
{ok, nil} | {error, runtime_error()}}.
cancel(Runtime, Key, Reason) ->
transition(Runtime, Key, <<"cancel"/utf8>>, fun(State) -> case State of
{queued, Seq} ->
{ok, {cancelled, Seq, Reason}};
{in_flight, Seq@1, _} ->
{ok, {cancelled, Seq@1, Reason}};
_ ->
{error,
{invalid_transition,
Key,
state_label(State),
<<"cancel"/utf8>>}}
end end).
-file("src/lightspeed/async/backpressure.gleam", 198).
?DOC(" Retry a failed/cancelled task by putting it back in queue.\n").
-spec retry(runtime(), binary()) -> {runtime(),
{ok, nil} | {error, runtime_error()}}.
retry(Runtime, Key) ->
transition(Runtime, Key, <<"retry"/utf8>>, fun(State) -> case State of
{failed, Seq, _} ->
{ok, {queued, Seq}};
{cancelled, Seq@1, _} ->
{ok, {queued, Seq@1}};
_ ->
{error,
{invalid_transition,
Key,
state_label(State),
<<"retry"/utf8>>}}
end end).
-file("src/lightspeed/async/backpressure.gleam", 438).
-spec find_state(list(task()), binary()) -> gleam@option:option(task_state()).
find_state(Tasks_rev, Key) ->
case Tasks_rev of
[] ->
none;
[Task | Rest] ->
case erlang:element(2, Task) =:= Key of
true ->
{some, erlang:element(3, Task)};
false ->
find_state(Rest, Key)
end
end.
-file("src/lightspeed/async/backpressure.gleam", 217).
?DOC(" Lookup task state by key.\n").
-spec state(runtime(), binary()) -> gleam@option:option(task_state()).
state(Runtime, Key) ->
find_state(erlang:element(4, Runtime), Key).
-file("src/lightspeed/async/backpressure.gleam", 222).
?DOC(" Boundary accessor.\n").
-spec runtime_boundary(runtime()) -> boundary().
runtime_boundary(Runtime) ->
erlang:element(2, Runtime).
-file("src/lightspeed/async/backpressure.gleam", 227).
?DOC(" In-flight task count.\n").
-spec in_flight_count(runtime()) -> integer().
in_flight_count(Runtime) ->
erlang:element(5, Runtime).
-file("src/lightspeed/async/backpressure.gleam", 232).
?DOC(" Queued task count.\n").
-spec queued_count(runtime()) -> integer().
queued_count(Runtime) ->
erlang:element(6, Runtime).
-file("src/lightspeed/async/backpressure.gleam", 237).
?DOC(" Stable adapter mode label.\n").
-spec mode_label(adapter_mode()) -> binary().
mode_label(Mode) ->
case Mode of
push_pull ->
<<"push_pull"/utf8>>;
push_only ->
<<"push_only"/utf8>>;
pull_only ->
<<"pull_only"/utf8>>
end.
-file("src/lightspeed/async/backpressure.gleam", 257).
?DOC(" Stable runtime-error label.\n").
-spec error_label(runtime_error()) -> binary().
error_label(Error) ->
case Error of
{duplicate_task, Key} ->
<<"duplicate_task:"/utf8, Key/binary>>;
{queue_saturated, Key@1, Max_queued} ->
<<<<<<"queue_saturated:"/utf8, Key@1/binary>>/binary, ":"/utf8>>/binary,
(erlang:integer_to_binary(Max_queued))/binary>>;
{unknown_task, Key@2} ->
<<"unknown_task:"/utf8, Key@2/binary>>;
{invalid_transition, Key@3, State, Action} ->
<<<<<<<<<<"invalid_transition:"/utf8, Key@3/binary>>/binary,
":"/utf8>>/binary,
State/binary>>/binary,
":"/utf8>>/binary,
Action/binary>>
end.
-file("src/lightspeed/async/backpressure.gleam", 449).
-spec task_signature(task_state()) -> binary().
task_signature(State) ->
case State of
{queued, Seq} ->
<<"queued:"/utf8, (erlang:integer_to_binary(Seq))/binary>>;
{in_flight, Seq@1, Started_ms} ->
<<<<<<"in_flight:"/utf8, (erlang:integer_to_binary(Seq@1))/binary>>/binary,
":"/utf8>>/binary,
(erlang:integer_to_binary(Started_ms))/binary>>;
{succeeded, Seq@2} ->
<<"succeeded:"/utf8, (erlang:integer_to_binary(Seq@2))/binary>>;
{failed, Seq@3, Reason} ->
<<<<<<"failed:"/utf8, (erlang:integer_to_binary(Seq@3))/binary>>/binary,
":"/utf8>>/binary,
Reason/binary>>;
{cancelled, Seq@4, Reason@1} ->
<<<<<<"cancelled:"/utf8, (erlang:integer_to_binary(Seq@4))/binary>>/binary,
":"/utf8>>/binary,
Reason@1/binary>>
end.
-file("src/lightspeed/async/backpressure.gleam", 281).
?DOC(" Tasks in stable sequence order.\n").
-spec tasks(runtime()) -> list(task_state()).
tasks(Runtime) ->
_pipe = erlang:element(4, Runtime),
_pipe@1 = lists:reverse(_pipe),
gleam@list:map(_pipe@1, fun(Task) -> erlang:element(3, Task) end).
-file("src/lightspeed/async/backpressure.gleam", 461).
-spec join_with(binary(), list(binary())) -> binary().
join_with(Separator, Values) ->
case Values of
[] ->
<<""/utf8>>;
[Value] ->
Value;
[Value@1 | Rest] ->
<<<<Value@1/binary, Separator/binary>>/binary,
(join_with(Separator, Rest))/binary>>
end.
-file("src/lightspeed/async/backpressure.gleam", 269).
?DOC(" Stable runtime signature for deterministic fixtures.\n").
-spec signature(runtime()) -> binary().
signature(Runtime) ->
<<<<<<<<<<<<<<"mode="/utf8,
(mode_label(
erlang:element(
2,
erlang:element(2, Runtime)
)
))/binary>>/binary,
"|in_flight="/utf8>>/binary,
(erlang:integer_to_binary(erlang:element(5, Runtime)))/binary>>/binary,
"|queued="/utf8>>/binary,
(erlang:integer_to_binary(erlang:element(6, Runtime)))/binary>>/binary,
"|tasks="/utf8>>/binary,
(join_with(
<<","/utf8>>,
gleam@list:map(tasks(Runtime), fun task_signature/1)
))/binary>>.
