This is part 3 of my series on building a harness in Erlang. To recap we have:
- An agent that calls the OpenAI responses API and runs tools, including bash
- A raw-mode terminal UI with slash commands
- A turn state machine that queues messages while the agent is busy
- Tools that run asynchronously, with timeouts and cleanup when they crash or hang
- An agent that pushes its state to every subscriber
The glaring flaw in what we have thus far is durability. If the agent dies, there’s no way of resuming it. One of the strange things about Erlang is that it has a lot of stuff built in that you usually need to reach for an external component in other languages. It has three related data storage mechanisms.
- ETS: an in memory store.
- DETS: ETS but persisted to disk.
- Mnesia: a transactional distributed database built on ETS.
Given we want to recover the history even if the whole machine dies, ETS being memory only isn’t a solution. DETS gives you persistence but would be limited to being run on a single node. Mnesia on the other hand would allow you to run multiple nodes each with a copy of the data so that if one goes down, the system can continue.
Mnesia does have a few rough edges though. Chief among them is the default behaviour after recovering from a network partition. It will let both sides of the split continue writing happily but when they reconnect you’ll get an inconsistent database error and it won’t automatically heal them for you. We can modify this behaviour by demanding strong consistency but it does come at a cost.
We could also use an external database, Postgres or SQLite but given the convenience of running Mnesia distributed, that’s where we’ll start.
Starting mnesia
Mnesia is an application in erlang. We need to add it to our harness.app.src file. Rebar3 shell will automatically start this for us.
{application,
harness,
{applications, [kernel, stdlib, ssl, inets, mnesia]},
%%%We point mnesia at a data directory in main with application:set_env(mnesia, dir, "./data"), then set it up when our application starts.
%harness_app.erl
start(_StartType, _StartArgs) ->
ok = setup_mnesia(),
ok = conversation:create_tables(),
ok = conversation:wait_for_tables(),
harness_sup:start_link().Mnesia writes to memory only by default. setup_mnesia changes this node’s schema to disc_copies so our tables can live on disk too.
Creating a table
Mnesia requires we use a slightly strange erlang feature known as records. Different from types, they are essentially tuples with named fields. Mnesia tables store these values. Modelling data as tuples isn’t strange, for me it’s the syntax and the sharing of these records in .hrl files that I’m not fond of. I’m going to try to quarantine these records to within their own conversation module. The interface will take and return maps rather than records.
%% conversation.erl
-record(entry, {key, item}).
-record(conversation, {id, next_seq, model_kind, instructions, base_commit, patch, owner}).
create_tables() ->
ok = created_or_exists(
mnesia:create_table(entry, [
{attributes, record_info(fields, entry)},
{type, ordered_set},
{disc_copies, [node()]},
{majority, true}
])
),
%%% the same again for the conversation table
ok.
% Messages should be sorted oldest to newest left to right
append(Id, Epoch, Messages) ->
case
mnesia:transaction(fun() ->
%%% read the conversation row into H
Node = node(),
case H#conversation.owner of
{Node, Epoch} -> ok;
_ -> mnesia:abort(fenced)
end,
%%% write each message as an entry keyed {Id, Seq}, then bump next_seq
end)
of
{atomic, Next} -> {ok, Next};
{aborted, Reason} -> {error, Reason}
end.
claim(Id) ->
case
mnesia:transaction(fun() ->
case mnesia:read(conversation, Id, write) of
[#conversation{owner = {_, Epoch}} = C] ->
New = Epoch + 1,
mnesia:write(C#conversation{owner = {node(), New}}),
New;
[] ->
mnesia:abort(not_found)
end
end)
of
{atomic, Epoch} -> {ok, Epoch};
{aborted, Reason} -> {error, Reason}
end.The most important aspect is in create_tables. They are an ordered_set which gives us the ability to traverse the collection in the order of its primary key. Secondly, the majority tag means that writes can only proceed with a majority in agreement. In this case however, we’re using {disc_copies, [node()]} to create the table so we only have one node. append and load do as they are named, but append requires the caller’s node to be the current owner and hold the current epoch to succeed.
How does one become the owner? Calling claim will bump the epoch and make the caller’s node the owner. The last node to call claim owns the conversation. This has potential for a few pitfalls. Particularly when combined with supervisor restarts, you could imagine a degenerate case where multiple workers fight over the same conversation id, stealing it from one another infinitely. For now we only care about preventing history being written to the same conversation by two different processes at the same time.
Wiring it up
The agent still has no notion of conversation id or epoch and it will need these to use our new conversation api. We need to add them to the agent state.
%% agent.erl
-type state(Turn) ::
#{
%%% as before
conv_id := binary(),
epoch := pos_integer() | unsaved
}.Now we have to decide how the agent will get its id. We need to be able to resume conversation so we must allow it to be passed in by a caller somehow. The right place for this is init. We could let the agent handle creating a new id when one isn’t supplied. I’m going to make the caller responsible for generating and passing in the id.
The agent will check if the id exists, then claim the conversation. Otherwise it will create a new conversation with said id.
The downside with that approach is that the creation of an agent process will create a conversation. Should we then quit we’ll have an empty conversation polluting our history. So we need to lazily create it on the first message. First we need to add another function in conversation.
-export([new_id/0, create/3]).
new_id() -> binary:encode_hex(crypto:strong_rand_bytes(8), lowercase).
create(Id, #{model := Model, instructions := Instructions}, Entries) ->
case
mnesia:transaction(fun() ->
case mnesia:read(conversation, Id, write) of
[] ->
%%% write Entries from seq 1, as in append
mnesia:write(#conversation{
id = Id,
model_kind = Model,
instructions = Instructions,
%%% ...
owner = {node(), 1}
}),
{Next, 1};
[_] ->
mnesia:abort(id_taken)
end
end)
of
{atomic, V} -> {ok, V};
{aborted, Reason} -> {error, Reason}
end.new_id is a convenience for generating the ids which we can leave to the conversation module to decide. create writes the conversation row and its first entries in a single transaction.
Now to modify our agent to use an id passed in.
%% agent.erl
init(#{conv_id := Id} = Opts) ->
%%% as before
State = #{
%%% as before
conv_id => Id,
epoch => unsaved
},
{ok, load_existing_conversation(State)}
end.
load_existing_conversation(#{conv_id := Id} = State) ->
case conversation:claim(Id) of
{ok, Epoch} ->
{ok, C, E} = conversation:load(Id),
Input = lists:reverse(E),
State#{
epoch => Epoch,
model => maps:get(model, C),
input => Input,
instructions => maps:get(instructions, C),
history => lists:append([lists:reverse(to_conv(N)) || N <- Input])
};
{error, not_found} ->
State;
{error, Reason} ->
exit(Reason)
end.
write_history(#{conv_id := Id, epoch := unsaved} = State, Entries) ->
{ok, {_, Epoch}} = conversation:create(
Id,
#{model => maps:get(model, State), instructions => maps:get(instructions, State)},
lists:reverse(Entries)
),
State#{epoch => Epoch};
write_history(#{conv_id := Id, epoch := Epoch} = State, Entries) ->
{ok, _} = conversation:append(Id, Epoch, lists:reverse(Entries)),
State.
%% Entries are newest first.
push_history(Entries, #{input := Input, history := History} = State) ->
NewState = write_history(State, Entries),
NewState#{
input => Entries ++ Input,
history => lists:append([lists:reverse(to_conv(E)) || E <- Entries]) ++ History
}.We now need a conv_id to create an agent. We also initialise the epoch to unsaved. It then attempts to load the conversation overwriting the default state, otherwise it’s unchanged.
push_history is the function through which all pushes to the State input and history occur. We have added a new call to a new function write_history which will create when epoch is unsaved (meaning the conversation is new) updating the epoch in the state, otherwise it will append. This way we only create conversations when a message is sent.
To test it, I hard-coded a conv_id into the agent’s child spec in harness_sup.
It works. It retains the conversation history and resumes where we left off. But what if we want more than just one conversation?
Dynamically launching agents
For convenience in our tui, we’ve hard coded the current agent to take the alias of {local, agent}. It makes it easy to call it but it means that our front end can only communicate with the process registered to that alias. Furthermore, this sole agent is launched at start up in the harness_sup. This works for just one agent. But we want the ability to launch new ones at will. So we’ll add another supervisor layer to assist.
agent_sup lets us start agents on demand.
%% agent_sup.erl
start_link() ->
supervisor:start_link({local, ?SERVER}, ?MODULE, []).
start_agent(Opts) -> supervisor:start_child(?MODULE, [Opts]).
init([]) ->
SupFlags =
#{
strategy => simple_one_for_one,
intensity => 5,
period => 10
},
ChildSpecs =
[
#{
id => agent,
start => {agent, start_link, []},
% Never restarted; whoever is watching reopens the conversation.
restart => temporary,
shutdown => 5000,
type => worker,
modules => [agent]
}
],
{ok, {SupFlags, ChildSpecs}}.The key takeaways here is that you can start agents via an exported start_agent function and the agents are temporary. The supervisor never restarts them, because the user can do this manually by reopening the session, so we are leaving it to the user for now.
Now we just amend harness_sup.erl to start the agent supervisor instead of the agent directly.
Lastly, we have been identifying our agent by a keyword agent. Given we have ambitions of running more than one agent at a time, we need a way to identify and address messages to each of our agents.
%% agent.erl
start_link(#{conv_id := Id} = Opts) ->
gen_server:start_link({via, global, {?MODULE, Id}}, ?MODULE, Opts, []).This change gives each agent a globally unique (within the erlang cluster) id of {agent, ConversationId}. This is convenient but there are some drawbacks:
- You need a global lock to write to it so it can become a bottleneck
- It assumes fully connected mesh so if there’s a network partition recovery is ugly
- The global registry needs to be replicated across every node
These are mostly performance concerns, my priority is correctness at present so the only one that bothers me is network partition behaviour. We can address that when we get to multinode setup.
Now that our agent has its own unique name, we need to update our TUI to store the current agent and only communicate or listen to that one.
main(Args) ->
%%% setup
case application:ensure_all_started(harness) of
{ok, _Started} ->
ConvId =
case Args of
[Id] -> list_to_binary(Id);
[] -> conversation:new_id()
end,
{ok, ModelPid} = create_model(ViewPid, ConvId),
%%%% start conversation
create_model(ViewPid, ConvId) ->
Pid = spawn_link(fun() ->
run_effect({open_conversation, ConvId}),
loop(ViewPid, #{conv => none})
end),
register(model, Pid),
{ok, Pid}.We can now supply an id for the conversation in our cli args and we supply that to model. The model then runs the effect open_conversation before looping.
What does it do?
run_effect({unqueue, AgentPid}) ->
gen_server:cast(AgentPid, {user, unqueue, self()});
run_effect({open_conversation, Id}) ->
Pid =
case agent_sup:start_agent(#{conv_id => Id}) of
{ok, New} -> New;
{error, {already_started, P}} -> P
end,
self() ! {agent_started, Id, Pid};
run_effect({subscribe, AgentPid}) ->
gen_server:cast(AgentPid, {subscribe, self()});
run_effect({send_agent_msg, AgentPid, Text}) ->
gen_server:cast(AgentPid, {user, message, self(), unicode:characters_to_binary(Text)}).
update({agent_started, Id, Pid}, State) ->
{State#{agent => Pid, conv_id => Id}, [{subscribe, Pid}]};
update({conv, AgentPid, NewConv}, #{conv := ConvState, agent := AgentPid} = State) ->
Conv =
case ConvState of
none -> NewConv;
OldConv -> maps:merge(OldConv, NewConv)
end,
{State#{conv => Conv}, []};
update({unqueued, AgentPid, []}, #{agent := AgentPid} = State) ->
{State, []};
update({unqueued, AgentPid, Unqueued}, #{agent := AgentPid} = State) ->
Taken = iolist_to_binary(lists:join(<<"\n">>, [M || {user, M} <- Unqueued])),
{State#{text => Taken}, []};The important thing to note here is the flow
- Open conversation and send message to self of agent started
- Update state to include Pid and conv Id and return subscribe effect
- Subscribe to agent
It’s slightly indirect, but it keeps the order of operations clean, update pure and no state changes in our effects. Were we to subscribe inside the open_conversation effect, we could receive messages and drop them or end up with invalid state. This version ensures that we only subscribe when our state has been updated.
starting and resuming a conversation
A new command
Now we can add a new command, the new command.
%%%
run_effect({unsubscribe, AgentPid}) ->
gen_server:cast(AgentPid, {unsubscribe, self()});
run_effect({open_conversation, new}) ->
run_effect({open_conversation, conversation:new_id()});
%%% run_effect({open_conversation, Id}) as before
command(<<"new">>, #{agent := AgentPid} = State) ->
{maps:remove(agent, State#{text => <<"">>, conv => none}), [
{unsubscribe, AgentPid}, {open_conversation, new}
]};
update({conv, _, _}, State) ->
{State, []};
update({unqueued, _, _}, State) ->
{State, []};We now remove the previous agent from our state, unsubscribe so we stop receiving messages we don’t need and open a new agent. We don’t kill the old agent so it will keep running after /new. We won’t address this yet, it actually needs careful design, blindly killing it on zero subscribers runs the risk of agents being killed prematurely.
We need those catch alls at the bottom because there is a period between changing our AgentPid and the unsubscribe being actioned by the old agent where we can still receive messages from the old agent. We don’t have a catch all update to crash on purpose when unhandled messages are received, therefore we need a matching clause or we would otherwise crash.
Orphaned function calls
Now we need to address a couple of issues. We aren’t shutting down gracefully, so some assistant messages are committed but never make it to disk if we quit right after they arrive. Secondly, when the assistant does return function calls and the response completes, the assistant message will end up in our database. However, if the agent stopped before the tools finished running there will be no corresponding result entry. If we then resume, the input is in a malformed state and the openai api will reject our request.
The assistant’s response is only added when we receive a response.completed message. I’m not that bothered by dropping in progress responses like we presently do. Tools that finish mid-stream get dropped too, since the call they answer hasn’t been saved yet. The danger is when the response is done but the tools are running. We need to cancel them, and every call that never got a result needs one saying it was interrupted.
gen_server gives us graceful shutdown behaviour. We just need to implement a terminate callback and trap exits in init. But terminate only runs if we get to shut down. A crash or a killed node skips it, so the interrupted results are better created when we load the conversation. That covers every case in one place.
%% agent.erl
-export([start_link/1, init/1, handle_cast/2, handle_call/3, handle_info/2, terminate/2]).
init(#{conv_id := Id} = Opts) ->
case os:getenv("OPENAI_API_KEY") of
false ->
{stop, missing_api_key};
Key ->
process_flag(trap_exit, true),
%%% initialise
push_history([], State) ->
State;
load_existing_conversation(#{conv_id := Id} = State) ->
%%% claim and load as before
Loaded = State#{
%%% ...
},
interrupt_unanswered(Loaded);
interrupt_unanswered(#{input := Input} = State) ->
Answered = [CallId || {tool_result, CallId, _} <- Input],
Calls = [
CallId
|| {assistant, Response} <- Input,
{function_call, CallId, _, _} <- openai:output(Response)
],
push_history(
[
{tool_result, CallId, <<"interrupted">>}
|| CallId <- Calls, not lists:member(CallId, Answered)
],
State
).
save_on_exit(_Reason, #{turn := #{phase := awaiting_tools, tool_results := Results}} = State) ->
push_history(Results, State),
ok;
save_on_exit(_Reason, State) ->
ok.
terminate(Reason, #{turn := Turn} = State) ->
cancel_tools(Turn),
save_on_exit(Reason, State),
ok.We have a new push_history clause so that nothing occurs when entries is []. We reuse cancel_tools, and save_on_exit keeps the results of any tools that did finish. Whatever is still unanswered gets its interrupted result the next time the conversation loads. If the agent crashes because the agent’s original push_history call was fenced, then push_history will fail a second time in terminate. Probably not ideal but I don’t think harmful at this stage.
Now we need to send the shutdown commands from main.
main(Args) ->
%%%
Reason = wait(Reader),
application:stop(harness),
application:stop(mnesia),That will stop our agent gracefully and when running on a single node as we are now, it will flush the writes that mnesia has committed only in memory onto disk. Unfortunately, we get no such control when the Erlang process (os process) dies so we cannot guarantee every completed message will be recoverable while running on one node.
Why?
That probably feels like overkill for such a simple application. We have supervisors, registries, pids and shutdowns going on. async python and two SQLite table would suffice. In fact, nothing we’ve done is better than what codex or claude code offer you. Saying it’s for fun explains why I’m doing it but it doesn’t explain why on earth I would find this fun in the first place.
I’m now going to start leveraging some of the things that Erlang makes uniquely easy to do. Running our application on multiple connected nodes.
First, our nodes each need names. main now parses its arguments with the built-in argparse module: a node name, an optional session to resume, and --headless to run without the TUI. Then it starts the node.
start_node(Name) ->
application:set_env(kernel, net_ticktime, 4),
{ok, _} = net_kernel:start(binary_to_atom(Name), #{name_domain => shortnames}),
application:set_env(mnesia, dir, "./data_" ++ binary_to_list(Name)),
%%% log to a file instead of the terminalThe important part to see is in start_node, we have set net_ticktime to 4. So if our nodes experience 4-5 seconds of silence from a peer, they consider it dead. We also write our data into data_<node> directory now so we don’t overlap.
Nodes abc
Our mnesia setup currently assumes a single node. What we instead want to do is to check if there’s a peer, if so copy their tables instead. If no peers then we will create the tables ourselves. This is not robust, if the timing is wrong you’ll end up with 2 nodes creating the tables. It’s also not great for restarting after a total outage, we need to get the restart order right. Namely the last node that went down needs to be the first that comes back.
%% harness_app.erl
start(_StartType, _StartArgs) ->
ok = setup_mnesia(),
ok = conversation:wait_for_tables(),
harness_sup:start_link().
stop(_State) ->
ok.
%% internal functions
setup_mnesia() ->
case [P || P <- peers(), net_adm:ping(P) =:= pong] of
[] ->
ok = disc_schema(),
conversation:create_tables();
Up ->
join(Up)
end.
peers() ->
{ok, Host} = inet:gethostname(),
[list_to_atom(N ++ "@" ++ Host) || N <- ["a", "b", "c"]] -- [node()].
join(Up) ->
{ok, _} = mnesia:change_config(extra_db_nodes, Up),
ok = disc_schema(),
lists:foreach(fun add_copy/1, [entry, conversation]).
add_copy(Table) ->
case mnesia:add_table_copy(Table, node(), disc_copies) of
{atomic, ok} -> ok;
{aborted, {already_exists, Table, _}} -> ok
end.
disc_schema() ->
case mnesia:change_table_copy_type(schema, node(), disc_copies) of
{atomic, ok} -> ok;
{aborted, {already_exists, schema, _, disc_copies}} -> ok;
Other -> {error, Other}
end.Our old setup_mnesia became disc_schema and is how we create the tables if a node is first. We’ve hard-coded our peer names a, b and c for now.
Starting nodes a, b and c in the background:
[michael@archlinux] ~/projects/erlang/harness ❯ harness c --headless &
[michael@archlinux] ~/projects/erlang/harness ❯ harness b --headless &
[michael@archlinux] ~/projects/erlang/harness ❯ harness a --headless &
[michael@archlinux] ~ ❯ erl -sname dbg -remsh a@archlinux
Erlang/OTP 29 [erts-17.0.1] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] [jit:ns]
Eshell V17.0.1 (press Ctrl+G to abort, type help(). for help)
(a@archlinux)1> mnesia:table_info(entry, disc_copies).
[a@archlinux,b@archlinux,c@archlinux]There it is, we have the entry table duplicated between our nodes. Because we have chosen majority, we need more than 50% of our copies to be online to progress. In our case of three, we can survive one node dying but no more than that.
Cleaning up after ourselves
Now that our conversation data is stored across all the nodes, we should probably create a mechanism for saying this agent is no longer running. We will do this by relinquishing ownership in the table. That will need some new functions in conversations.
release(Id, Epoch) ->
case
mnesia:transaction(fun() ->
Node = node(),
case mnesia:read(conversation, Id, write) of
[#conversation{owner = {Node, Epoch}} = C] ->
mnesia:write(C#conversation{owner = {none, Epoch}}),
Epoch;
[_] ->
mnesia:abort(not_owner);
[] ->
mnesia:abort(not_found)
end
end)
of
{atomic, T} -> {ok, T};
{aborted, Reason} -> {error, Reason}
end.
held() ->
mnesia:dirty_select(
conversation,
[
{#conversation{id = '$1', owner = {'$2', '_'}, _ = '_'}, [{'=/=', '$2', none}], [
{{'$1', '$2'}}
]}
]
).release sets the owner to {none, Epoch} if the node calling it is the owner and the caller has the right epoch. Otherwise it rejects the release. held on the other hand will give us all the agents currently owned. Why do we need that?
If we make our agents release ownership on graceful shutdown. Any agent with an owner that is not currently online was not brought down gracefully. If you combine this with a notification that a node went down you can find what work was running on that node.
Now we wire up the agent to release a conversation during shutdown.
%% agent.erl
save_on_exit(_Reason, #{turn := #{phase := awaiting_tools, tool_results := Results}} = State) ->
release_ownership(push_history(Results, State)),
ok;
save_on_exit(_Reason, State) ->
release_ownership(State),
ok.
release_ownership(#{conv_id := Id, epoch := Epoch}) ->
conversation:release(Id, Epoch).
unfinished([{assistant, _} | _]) -> false;
unfinished([_ | _]) -> true;
unfinished([]) -> false.The unfinished function will tell us whether a conversation ended while in progress or died cleanly. A clean death for the agent is one where it was in idle state, the response completed and there were no tools running. In which case the last thing written will be the assistant response. We’ve already solved the issue of in progress tool calls, so we can leverage that here.
%% agent.erl
init(#{conv_id := Id} = Opts) ->
case os:getenv("OPENAI_API_KEY") of
false ->
{stop, missing_api_key};
Key ->
process_flag(trap_exit, true),
State = #{
%%% as before
},
NewState = load_existing_conversation(State),
case unfinished(maps:get(input, NewState)) of
true ->
gen_server:cast(self(), resume),
{ok, NewState};
false ->
case maps:get(revive, Opts, false) of
true ->
release_ownership(NewState),
ignore;
false ->
{ok, NewState}
end
end
end.
cast(resume, #{turn := idle} = State) ->
call_llm(State);If we find ourselves in an unfinished state we cast a resume message to ourselves. This tells the agent to immediately call the llm to continue. This is not an obvious choice, we are automatically retrying something that failed. At the very least it could be wasteful and costly, in the worst case we could experience cascading failures were that agent the thing that brought the node down and left it unfinished.
If the agent did finish, we check a new revive flag in init. This says whether the agent is being revived after a failure. If it’s being revived and not unfinished we release ownership and stop the agent. Otherwise it’s a resumption directed by the user in which case we proceed as normal.
Durable agents
Now we can move onto the interesting piece. We are going to try to make our agents durable. That is, if I start a session/job on my computer and it’s part of a mesh, I want it to automatically continue on another machine.
In order to do that we need a way of detecting that nodes are down. We’ll create a new gen_server named reviver. Its job is to revive agents who died while working.
%% reviver.erl
-module(reviver).
-behaviour(gen_server).
-export([init/1, start_link/0, handle_cast/2, handle_info/2, handle_call/3]).
start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
init(_) ->
% get notifications if nodes die
{ok, _} = mnesia:subscribe(system),
lists:foreach(fun revive/1, [
Id
|| {Id, Owner} <- conversation:held(), not lists:member(Owner, nodes())
]),
{ok, #{}}.
handle_info({mnesia_system_event, {mnesia_down, Node}}, State) ->
case hd(lists:sort([node() | mnesia:system_info(running_db_nodes)])) =:= node() of
true -> [revive(Id) || {Id, Owner} <- conversation:held(), Owner =:= Node];
false -> ok
end,
{noreply, State};
handle_info({retry, Id}, State) ->
revive(Id),
{noreply, State};
handle_info(_, State) ->
{noreply, State}.
revive(Id) ->
case agent_sup:start_agent(#{conv_id => Id, revive => true}) of
{ok, _} ->
ok;
{error, {already_started, Pid}} ->
case lists:member(node(Pid), mnesia:system_info(running_db_nodes)) of
true -> ok;
_ -> erlang:send_after(500, self(), {retry, Id})
end;
{error, Reason} ->
logger:warning("could not revive ~ts: ~p", [Id, Reason])
end.
handle_cast(_, State) -> {noreply, State}.
handle_call(_, _, State) -> {reply, ok, State}.By subscribing to system events for mnesia our reviver can find out when a node disconnects. The reviver has one job - restarting potentially unfinished agents. It does this when an mnesia node disconnects, looking only at the disconnected nodes agents, and it does so at start-up where it looks at any agent held by its own node or a currently disconnected node. We need a retry because mnesia and our global registry can disagree and mnesia is the source of truth, so when global says the pid exists but mnesia says that node is dead we listen to mnesia and try again later.
Another thing to note is the way of choosing which node is tasked with restarting the dead agent. Any are eligible so how do we decide? We choose based on alphabetical order of node names. This is quite brittle because if the chosen node dies while restarting an agent, another node won’t restart it in its stead. It will have to wait for a node to restart (or at least its reviver). If the nodes disagree on who is in the cluster like during a network partition more than one could attempt to restart the dead agent.
These are classic distributed systems issues. Fixing them is involved and beyond the scope of this post.
Let’s start our reviver automatically by adding it to our root supervisor.
%% harness_sup.erl
init([]) ->
SupFlags =
#{
strategy => rest_for_one,
intensity => 3,
period => 5
},
ChildSpecs =
[
%%% agent_sup as before
#{
id => reviver,
start => {reviver, start_link, []},
restart => permanent,
shutdown => 5000,
type => worker,
modules => [reviver]
}
],
{ok, {SupFlags, ChildSpecs}}.The change that matters is the new strategy of rest_for_one. That means that the supervisor will use the start-up order (order in childspec list) to determine what to restart on a crash. In our case if the agent_sup dies, our reviver will restart too. This is useful because reviver is useless without a way of spawning agents.
Two new commands
We will be creating /list to show the active agents running and where and /resume to hook up to a running agent.
run_effect(list_agents) ->
self() ! {agents, list_agents()}.
update({agents, Agents}, State) ->
{State#{agents => Agents}, []};
render_agents(none) ->
[];
render_agents([]) ->
<<"No agents\r\n">>;
render_agents(Agents) ->
[
<<"Agents:\r\n">>,
[[<<" ">>, Id, <<" ">>, agent_status(Status), <<"\r\n">>] || {Id, Status} <- Agents]
].
agent_status({running, Node}) ->
[<<"running on ">>, atom_to_binary(Node)];
agent_status({interrupted, Node}) ->
[<<"interrupted (was on ">>, atom_to_binary(Node), <<")">>].
list_agents() ->
[
{Id,
case global:whereis_name({agent, Id}) of
Pid when is_pid(Pid) -> {running, node(Pid)};
undefined -> {interrupted, Owner}
end}
|| {Id, Owner} <- conversation:held()
].
command(<<"resume ", Id/binary>>, #{agent := AgentPid} = State) ->
{maps:remove(agent, State#{text => <<"">>, conv => none}), [
{unsubscribe, AgentPid}, {open_conversation, Id}
]};
command(<<"list">>, State) ->
{State#{text => <<"">>}, [list_agents]};/resume takes everything after ‘/resume ’ as the session id. Right now we don’t require that the id actually exists. It unsubscribes from the existing session too. /list shows all the running agents.
Running it all together now.
killing the node an agent is running on, and finding it running on another
The video shows three nodes, a, b and c, running in a cluster. We start a session ‘mysess’ in b. Ask it to run a multi-step job and then kill it with ctrl-c. The death of the node causes ‘mysess’ to migrate to the bottom terminal node and continue. Then I am able to restart the node b and hook into the session even while it takes place on node a.
If you looked carefully, I actually started the a node in a different directory. It ran in the ./stuff directory with several .txt files. b was in the parent directory. My command to the agent to cleanup the .txt files started in the parent but completed in the stuff directory. This is one of the most dangerous aspects of this system at present and something we will address in a later chapter. I had added a small directive to our instructions below but it was ultimately in vain as the agent lacks context around where it is. If you’ve been wondering what base_commit and patch were for in our record definition. They will help address this.
%% agent.erl
%% init(Opts)->
instructions =>
<<
(maps:get(
instructions,
Opts,
<<
"You are a personal agent. Assist your users with whatever they "
"ask you to do"
>>
))/binary,
"\n You are running in a distributed multinode setup. When you crash you can be revived anywhere. You should not assume you are running in the same directory or environment. You should check. Interrupted commands are not safe to retry without verification unless they are idempotent."
>>Some other issues we will have to address:
- We don’t have any concept of liveness, if we kill the system in progress and bring it back up in a week, it might restart a plan my groceries task long after it was ever needed.
- A job that kills its node can kill the next one too. If the job itself crashes the node, say a command that eats all the memory, the reviver moves it to b, b dies, it moves to c… and the whole cluster goes down one node at a time.
- Dead vs disconnected is the perennial challenge in distributed systems, we aren’t actually sure the node is dead. If a is only cut off from the others, b revives the job while a is still running it. Fencing stops a’s agent from writing history, but it doesn’t stop a’s bash commands. Both copies act on the world until a’s next write fails. Because the write fails, we don’t actually even have a log of what a did when it was alone.
- bash has no limits at all, tools can run many times and as the video showed with no regard for where the node is.
- Per the video, we need to start the nodes in a particular order or we run into issues with table setup.
These are just a few of the problems. Join me next time when I try to expose even more of them.