anchor.diff:  notify_watchers(Watchers, NodeRef, Current, New) ->
       F = fun(Watcher) -> Watcher ! {state_change, NodeRef, Current, New} end,
       [F(Watcher) || {Watcher, _Monitor} <- Watchers].
  +
  +
  +%%
  +%% Message processing and parsing code; this executes in the caller's context
  +%%
  +dispatch_raw_message([Addr, _Header, Body] = Message) ->
  +    Pid = case pushy_node_state_sup:get_process(Addr) of
  +              P when is_pid(P) ->
  +                  P;
  +              _Error ->
  +                  %% Nothing with that address found; we need to parse the body of the
  +                  %% message to figure out where to route it...
  +                  %%
  +                  %% NOTE: We need to use some care when creating FSMs. Ideally we would
  +                  %% verify the message before we create the FSM. However, this adds some
  +                  %% delay to the node startup path, and serializes things in an unpleasant
  +                  %% way. Instead we speculatively create a new FSM for the node_ref/addr
  +                  %% pair if there aren't any FSMs registered for either.
  +                  %%
  +                  %% So this is only an issue if there is no registered FSM for this
  +                  %% node_ref or address.
  +                  %%
  +                  %% If we get spoofed into creating an FSM, it will simply parse and drop
  +                  %% spoofed packets, and eventually go down because it hasn't received a
  +                  %% valid heartbeat. If the real node attempts to connect, it will cause
  +                  %% the FSM to change its registered address, and the spoofed packets
  +                  %% will continue to be rejected.
  +                  %%
  +                  %% The main threat is some DOS attack where we get tricked into creating
  +                  %% large numbers of FSMs, which would consume resources.
  +                  %%
  +                  %% One defense would be to require that we have a hmac key assigned for
  +                  %% this node: that would restrict the number of spoofable FSM's to the set
  +                  %% of nodes that have already made a validated call to the configuration
  +                  %% endpoint. A further refinement would be to make the node ref cheaper to
  +                  %% parse and extract (bypassing json perhaps?)
  +                  EJSon = jiffy:decode(Body),
  +                  NodeRef = get_node_ref(EJSon),
  +                  pushy_node_state_sup:get_or_create_process(NodeRef, Addr)
  +          end,
  +    Pid ! {raw_message, Message }.
  +
  +%%
  +%% This occurs in the fsm context
  +%%
  +maybe_process_and_dispatch_message(CurrentState, State, Message) ->
  +    case process_and_dispatch_message(Message, State) of
  +        {ok, State1} -> {next_state, CurrentState, State1};
  +        {should_die, State1} -> {stop, state_transition(CurrentState, shutdown, State1), State1}
  +    end.
  +
  +process_and_dispatch_message([Address, Header, Body], State) ->
  +    KeyFetch = fun key_fetch/2,
  +    State1 = try ?TIME_IT(pushy_messaging, parse_message, (Address, Header, Body, KeyFetch)) of
  +                 {ok, #pushy_message{} = Msg} ->
  +                     {ok, process_message(State, Msg)};
  +                 {error, #pushy_message{validated=bad_sig}} ->
  +                     lager:error("Command message failed verification: header=~s", [Header]),
  +                     {ok, State}
  +             catch
  +                 error:Error ->
  +                     Stack = erlang:get_stacktrace(),
  +                     lager:error("Command message parser failed horribly: header=~w~nstack~s", [Error, Stack]),
  +                     {ok, State};
  +                 Error ->
  +                     Stack = erlang:get_stacktrace(),
  +                     lager:error("Command message parser failed horribly: header=~w~nstack~s", [Error, Stack]),
  +                     {ok, State}
  +             end,
  +    State1.
  +
  +process_message(#state{node_ref=NodeRef} = State, #pushy_message{address=Address, body=Data}) ->
  +    lager:debug("Received message for Node ~p (address ~p)", [NodeRef, Address]),
  +    JobId = ej:get({<<"job_id">>}, Data),
  +    Type = ej:get({<<"type">>}, Data),
  +    send_node_event(JobId, NodeRef, Type),
  +    State.
  +
  +send_node_event(null, NodeRef, <<"heartbeat">>) ->
  +    lager:debug("Received heartbeat for node ~p with NULL job id", [NodeRef]),
  +    self() ! heartbeat;
  +send_node_event(JobId, NodeRef, <<"heartbeat">>) ->
  +    lager:debug("Received heartbeat for node ~p with job id ~p", [NodeRef, JobId]),
  +    case pushy_job_state_sup:get_process(JobId) of
  +        not_found -> pushy_node_state:rehab(NodeRef);
  +        _ -> noop
  +    end,
  +    pushy_node_state:heartbeat(NodeRef);
  +send_node_event(JobId, NodeRef, <<"ack_commit">>) ->
  +    pushy_job_state:node_ack_commit(JobId, NodeRef);
  +send_node_event(JobId, NodeRef, <<"nack_commit">>) ->
  +    pushy_job_state:node_nack_commit(JobId, NodeRef);
  +send_node_event(JobId, NodeRef, <<"ack_run">>) ->
  +    pushy_job_state:node_ack_run(JobId, NodeRef);
  +send_node_event(JobId, NodeRef, <<"nack_run">>) ->
  +    pushy_job_state:node_nack_run(JobId, NodeRef);
  +send_node_event(JobId, NodeRef, <<"complete">>)->
  +    pushy_job_state:node_complete(JobId, NodeRef);
  +send_node_event(null, NodeRef, <<"aborted">>) ->
  +    pushy_node_state:aborted(NodeRef);
  +send_node_event(JobId, NodeRef, <<"aborted">>) ->
  +    pushy_node_state:aborted(NodeRef),
  +    pushy_job_state:node_aborted(JobId, NodeRef);
  +send_node_event(JobId, NodeRef, undefined) ->
  +    lager:error("Status message for job ~p and node ~p was missing type field!~n", [JobId, NodeRef]);
  +send_node_event(JobId, NodeRef, UnknownType) ->
  +    lager:error("Status message for job ~p and node ~p had unknown type",
  +                [JobId, NodeRef, UnknownType]).
  +
  +get_node_ref(Data) ->
  +    %% This essentially debug code.
  +    _ClientName = ej:get({<<"client">>}, Data),
  +    OrgName  = ej:get({<<"org">>}, Data),
  +    NodeName = ej:get({<<"node">>}, Data),
  +    %% TODO: Clean up usage and propagation of org name vs org guid (OC-4351)
  +    OrgId = pushy_object:fetch_org_id(OrgName),
  +    {OrgId, NodeName}.
  +
  +
  +get_key_for_method(hmac_sha256, {_,_} = NodeRef) ->
  +    {hmac_sha256, Key} = pushy_key_manager:get_key(NodeRef),
  +    {ok, Key};
  +get_key_for_method(rsa2048_sha1, _) ->
  +    chef_keyring:get_key(pushy_priv).
  +
  +key_fetch(Method, EJson) ->
  +    NodeRef = get_node_ref(EJson),
  +    get_key_for_method(Method, NodeRef).
  
positive.code: handle_info(rehab_again, rehab, State) ->
      State1 = force_abort(State),
      {next_state, rehab, State1};
  handle_info({'DOWN', _MRef, _Type, Pid, _Reason}, StateName, #state{watchers=Watchers}=State) ->
      case lists:keytake(Pid, 1, Watchers) of
          false ->
  
negative.code:          terminate/3,
           code_change/4]).
  
  start_link({_,_} = NodeRef, NodeAddr) ->
      gen_fsm:start_link(?MODULE, [NodeRef, NodeAddr], []).
  
  heartbeat(NodeRef) ->
      send_info(NodeRef, heartbeat),
      ok.
  recv_msg(Message) ->
      dispatch_raw_message(Message).
  
  
  status(NodeRef) ->
      case call(NodeRef, current_state) of
  