anchor.diff:              loop(handle_watch(State, Watch));
           #unwatch{}=Unwatch ->
               loop(handle_unwatch(State, Unwatch));
  -        #lock_transaction{session=Session} ->
  -            loop(handle_lock_transaction(State, Session));
  -        #rollback_transaction{session=Session} ->
  -            loop(handle_rollback_transaction(State, Session));
  -        #commit_transaction{session=Session} ->
  -            loop(handle_commit_transaction(State, Session));
  -        #store_result{id=Session, result=Result} ->
  -            loop(handle_store_result(State, Session, Result));
  +        #lock_transaction{txn_id=TransactionId} ->
  +            loop(handle_lock_transaction(State, TransactionId));
  +        #rollback_transaction{txn_id=TransactionId} ->
  +            loop(handle_rollback_transaction(State, TransactionId));
  +        #commit_transaction{txn_id=TransactionId} ->
  +            loop(handle_commit_transaction(State, TransactionId));
  +        #store_result{id=Id, result=Result} ->
  +            loop(handle_store_result(State, Id, Result));
           stop ->
               io:format("storage bucket halting after receiving stop message~n");
           Any ->
  
positive.code:     redis_store:delete(Store, Key).
  
  handle_operation(Store, Session, #get{key=Key}, _Current) ->
      redis_store:get(Session, Store, Key);
  handle_operation(Store, Session, #set{key=Key, value=Value}, Current) ->
      update_meta(Key, Current),
      redis_store:set(Session, Store, Key, Value);
  handle_operation(Store, Session, #delete{key=Key}, Current) ->
      delete_meta(Key, Current),
      redis_store:delete(Session, Store, Key).
  
  add_operation(error, Operation) ->
      add_operation({ok, #transaction{}}, Operation);
  add_operation({ok, #transaction{}=Transaction}, Operation) ->
      Transaction#transaction{operations=[Operation|Transaction#transaction.operations]}.
  
  handle_transact(#state{transactions=Transactions}=State, #transact{session=Session, operation_id=Id, operation=Operation}) ->
      Session ! {self(), stored},
      NewTransaction = add_operation(dict:find(Session, Transactions), {Id, Operation}),
      State#state{transactions=dict:store(Session, NewTransaction, Transactions)}.
  
  handle_watch(#state{transactions=Transactions}=State, #watch{session=Session, key=Key}) ->
      Session ! {self(), ok},
      NewTransaction = add_watch(dict:find(Session, Transactions), Key),
      State#state{transactions=dict:store(Session, NewTransaction, Transactions)}.
  
  add_watch(error, Key) ->
      add_watch({ok, #transaction{}}, Key);
  add_watch({ok, #transaction{watches=Watches}=Transaction}, Key) ->
      Current = get(Key),
      put(Key, key_meta:watch(Current)),
      Transaction#transaction{watches=[{Key, key_meta:version(Current)}|Watches]}.
  
  handle_unwatch(#state{transactions=Transactions}=State, #unwatch{session=Session}) ->
      Session ! {self(), ok},
      State#state{transactions=remove_watch(Transactions, Session, dict:find(Session, Transactions))}.
  
  remove_watch(Transactions, _Session, error) ->
      Transactions;
  remove_watch(Transactions, Session, {ok, Transaction}) ->
      remove_watches(Transaction#transaction.watches),
      dict:store(Session, Transaction#transaction{watches=[]}, Transactions).
  
  remove_watches([]) ->
      ok;
  
negative.code: 
  -record(state, {transactions, store}).
  -record(transaction, {session, operations=[], deferred=[], watches=[], locked=false}).
  
  -record(single_operation, {session}).
  -record(transaction_operation, {txn_id}).
  
  start() ->
      io:format("starting storage bucket with pid ~p~n", [self()]),
  