anchor.diff:                      erlang:error({configuration, Application, OptionName, Value})
               end;
           undefined ->
  -            error_logger:error_msg("Improper Configuration: ~p / ~p was undefined!~n", [Application, OptionName]),
  +            error_logger:error_msg("Improper Configuration: ~p / ~p was undefined!~n",
  +                                   [Application, OptionName]),
               erlang:error({configuration, Application, OptionName, undefined})
       end.
   
  -%% @doc Delete each checksummed file in S3
  -delete(OrgId, Checksums) when is_list(Checksums),
  -                              is_binary(OrgId) ->
  -
  +
  +-type request_fun() :: check_file | delete_file.
  +
  +%% @doc Parallelizes HTTP requests to S3.
  +%%
  +%% `Operation' is an atom naming a function in this module that will be used to actually
  +%% make the HTTP requests. The literal atom is also used for informative error logging.
  +%%
  +%% Returns a tuple containing a list of checksums for which the operation was successful, a
  +%% list of checksums that were not found, the number of timeouts that occurred (see
  +%% `timeout/0'), and the number of other unspecified errors that occurred.
  +-spec parallelize(OrgId :: object_id(),
  +                  Checksums :: [binary()],
  +                  Operation :: request_fun()) ->
  +                         {OkChecksums :: [binary()],
  +                          MissingChecksums :: [binary()],
  +                          NumTimeouts :: non_neg_integer(),
  +                          NumErrors :: non_neg_integer()}.
  +parallelize(OrgId, Checksums, Operation) -> %%, RequestFun) ->
  + 
       Bucket = chef_s3:bucket(),
       AwsConfig = chef_s3:get_config(),
       Timeout = timeout(),
   
  -    %% Using erlware_commons' ftmap ("fault-tolerant map") to handle the gory details of
  -    %% chunking and parallelizing the work of deleting files from S3 / Bookshelf.
  -    %%
  -    %% A 'malt' is erlware_commons jargon for the configuration 'blob' that specifies how an
  -    %% operation is to be parallelized using the ec_plist module (my favorite option for its
  -    %% meaning is "Malt is A List Tearing Specification"; see the module docs for more
  -    %% details on the various options available)
  -    %%
  +    %% Create a function to map across the list of checksums.  It spawns an additional
  +    %% process to perform the request in order to control timeout situations.
  +    %% 
  +    %% This is necessary because we'd like to keep track of individual operation information
  +    %% like this, but erlware_commons now obscures timeout information.  Additionally,
  +    %% erlware_commons appears to only allow clients to specify a timeout on an entire list
  +    %% operation as a whole, instead of a timeout on each individual list item operation.
  +    %%
  +    %% By spawning a separate process and managing the timeout ourselves, we can once again
  +    %% capture this information.  Additionally, we no longer need to specify a timeout via the
  +    %% ec_plist "malt" configuration, since that now basically takes care of itself.
  +    MapFun = fun(Checksum) ->
  +                     Me = self(),
  +    
  +                     %% This token is used below in the receive block (just look!) to make
  +                     %% absolutely certain that we are only processing the exact message we
  +                     %% are expecting, as opposed to any other messages the receiving
  +                     %% process may be getting from elsewhere in the system.
  +                     %%
  +                     %% It's admittedly a bit of paranoia, but should help insulate from
  +                     %% potential future changes in erlware_commons (that involve message
  +                     %% passing).
  +                     %%
  +                     %% Also, paranoia.
  +                     Token = erlang:make_ref(),
  +                     Worker = proc_lib:spawn_link(fun() ->
  +                                                          %% Figure out what private function we should use
  +                                                          OpFun = get_fun(Operation),
  +                                                          Result = OpFun(OrgId,
  +                                                                         AwsConfig,
  +                                                                         Bucket,
  +                                                                         Checksum),
  +                                                          Me ! {Token, Result, self()}
  +                                                  end),
  +                     receive
  +                         {Token, Response, Worker} ->
  +                             Response
  +                     after Timeout ->
  +                             error_logger:error_msg("Operation '~p' on checksum: ~p for org ~p from bucket ~p has taken longer than ~p ms; killing spawned worker process ~p~n",
  +                                                    [Operation, Checksum, OrgId, Bucket, Timeout, Worker]),
  +                             erlang:exit(Worker, kill),
  +                             {timeout, Checksum}
  +                     end
  +             end,
  +    
       %% Since making an HTTP request is a high-latency, IO-bound operation, we use a
       %% configuration where each list chunk has a single item, but we have multiple processes
       %% processing chunks.  This prevents pathological situations (like, for example, a large
       %% chunk of requests that all timeout taking LENGTH * TIMEOUT ms to complete), while
       %% maximizing the overall throughput.
       %%
  -    %% This configuration will also allow there to be fanout() HTTP requests in flight at
  +    %% This configuration will also allow there to be `fanout()' HTTP requests in flight at
       %% any given time, which makes for more even throughput.
       %%
  -    %% While we can specify a {timeout, Millis} tuple in the malt, this only applies to how
  -    %% long it takes to process the entire list; there does not currently appear to be a
  -    %% built-in way to manage individual process timeouts.  To circumvent this, we'll spawn
  -    %% helper processes to manage timeouts ourselves (see comments for spawn_deleter/5
  -    %% for more details).
  -    Malt = [1,                      %% Number of items to process in a given batch
  -            {processes, fanout()}], %% Have fanout() processes working on the entire list
  -
  -    Results = ec_plists:ftmap(fun(Checksum) ->
  -                                      spawn_deleter(OrgId, AwsConfig, Timeout, Bucket, Checksum)
  -                              end,
  -                              Checksums,
  -                              Malt),
  -
  -    %% The results will be a list of lists, so we must flatten it.  We then process the
  -    %% flattened results, splitting them into successfully-deleted checksums, missing
  -    %% checksums (i.e., we got a 404 when trying to delete them), timeout count and error
  -    %% count.
  -    %%
  -    %% Note that ec_plists:ftmap/3 wraps each resulting list item in a {value, X} tuple.  I
  -    %% think this is a hold-over from a previous implementation that doesn't really make
  -    %% much sense anymore, but whatever...
  -    {Ok, Missing, NumTimeouts, NumErrors} =
  +    %% While we can also specify a `{timeout, Millis}' tuple, this only applies to how long
  +    %% it takes to process the entire list; there does not currently appear to be a built-in
  +    %% way to manage individual process timeouts. This is why we manage our own timeouts, as
  +    %% implemented above in `MapFun'.
  +    %%
  +    %% See the documentation for erlware_commons' ec_plists module for more details.
  +    ParallelConfig = [1, {processes, fanout()}],
  +
  +    %% Now we actually get to perform the requests!
  +    Results = ec_plists:ftmap(MapFun, Checksums, ParallelConfig),
  +    
  +    %% Now we need to consolidate our results.
  +    %%
  +    %% Items in `Results' are each wrapped in a `{value, Term}' tuple (this is an
  +    %% erlware_commons thing).  I think this wrapping is a hold-over from a previous
  +    %% implementation of erlware_commons that doesn't really make much sense anymore, but
  +    %% whatever...
  +    {OkChecksums, MissingChecksums, NumTimeouts, NumErrors} =
           lists:foldl(fun({value, Result}, {Ok, Missing, Timeouts, Errors}) ->
                               case Result of
                                   {ok, Checksum} -> {[Checksum | Ok], Missing, Timeouts, Errors};
  
positive.code:                             end
                      end,
                      {[], [], 0, 0},
                      lists:flatten(Results)),
  
      %% A final bit of processing to sort the lists of checksums before we return
      {{ok, lists:sort(Ok)},
       {missing, lists:sort(Missing)},
       {timeout, NumTimeouts},
       {error, NumErrors}}.
  
  %% @doc Delete an existing checksum file in S3
  -spec delete_file(OrgId :: object_id(),
                    AwsConfig :: mini_s3:config(),
                    Bucket :: string(),
                    Checksum :: binary()) -> {'error', binary()} |
                                             {'missing', binary()} |
                                             {'ok', binary()}.
  delete_file(OrgId, AwsConfig, Bucket, Checksum) ->
      Key = chef_s3:make_key(OrgId, Checksum),
  
      try mini_s3:delete_object(Bucket, Key, AwsConfig) of
          %% Return value doesn't much matter here but for documentation's sake it looks like:
          %%
          %%  [{delete_marker, list_to_existing_atom(Marker)}, {version_id, Id}]
          %%
          %% The main take-away here is that the call to delete_object/3 was successful and
          %% didn't throw an error
          _Response ->
              {ok, Checksum}
      catch
          error:{aws_error, {http_error,404,_}} ->
              %% We got a 404.  No biggie; mark it as 'missing' and move on
              error_logger:warning_msg("Deletion of file (checksum: ~p) for org ~p from bucket ~p (key: ~p) failed because the file was not found~n",
                                       [Checksum, OrgId, Bucket, Key]),
              {missing, Checksum};
          ExceptionClass:Reason->
              %% Something unanticipated happened.  We should log the specific reason for
              %% later analysis, but as far as the overall deletion operation is concerned,
              %% this is "just an error", and we can continue along.
              error_logger:error_msg("Deletion of file (checksum: ~p) for org ~p from bucket ~p (key: ~p) raised exception ~p:~p~n",
                                     [Checksum, OrgId, Bucket, Key, ExceptionClass, Reason]),
              {error, Checksum}
      end.
  
  %% @doc Spawns a worker process to perform the actual deletion of an object from S3 /
  %% Bookshelf so that we can impose a timeout on the operation.
  %%
  %% This is necessary because we'd like to keep track of individual operation information
  %% like this, but erlware_commons now obscures timeout information.  Additionally,
  %% erlware_commons appears to only allow clients to specify a timeout on an entire list
  %% operation as a whole, instead of a timeout on each individual list item operation.
  %%
  %% By spawning a separate process and managing the timeout ourselves, we can once again
  %% capture this information.  Additionally, we no longer need to specify a timeout via the
  %% ec_plist "malt" configuration, since that now basically takes care of itself.
  -spec spawn_deleter(OrgId :: object_id(),
                      AwsConfig :: mini_s3:config(),
                      Timeout :: integer(),
                      Bucket::string(),
                      Checksum :: binary()) -> {'error', binary()} |
                                               {'missing', binary()} |
                                               {'ok', binary()} |
                                               {'timeout', binary()}.
  spawn_deleter(OrgId, AwsConfig, Timeout, Bucket, Checksum) ->
      Me = self(),
  
      %% This token is used below in the receive block (just look!) to make absolutely certain
      %% that we are only processing the exact message we are expecting, as opposed to any
      %% other messages the receiving process may be getting from elsewhere in the system.
      %%
      %% It's admittedly a bit of paranoia, but should help insulate from potential future
      %% changes in erlware_commons (that involve message passing).
      %%
      %% Also, paranoia.
      Token = erlang:make_ref(),
  
      Worker = proc_lib:spawn_link(fun() ->
                                           Result = delete_file(OrgId, AwsConfig, Bucket, Checksum),
                                           Me ! {Token, Result, self()}
                                   end),
      receive
          {Token, Response, Worker} ->
              Response
      after Timeout ->
              error_logger:error_msg("Deletion of file (checksum: ~p) for org ~p from bucket ~p has taken longer than ~p ms; killing spawned worker process ~p~n",
                                     [Checksum, OrgId, Bucket, Timeout, Worker]),
              erlang:exit(Worker, kill),
              {timeout, Checksum}
      end.
  
  %% File Checking Code Below
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  
  -record(state, {
            token :: reference(),
            checksums = [] :: [binary()],
            org_id :: object_id(),
            bucket :: string(),
            count = 0 :: non_neg_integer(),
            workers = [] :: [pid()],
  
            ok = [] :: [binary()],
            missing = [] :: [binary()],
            errors = 0 :: non_neg_integer()
           }).
  
  
  
  %% @doc Verify that each checksummed file is stored in S3 by checking its metadata
  fetch_md(OrgId, Checksums) when is_list(Checksums),
                                  is_binary(OrgId) ->
      parallel_fetch(OrgId, Checksums).
  
  parallel_fetch(OrgId, Checksums) ->
  
      Bucket = chef_s3:bucket(),
  
      Remainder = length(Checksums) rem ?MAX,
      Token = erlang:make_ref(),
  
      % Using a record since there's quite a bit to keep track of;
      % otherwise the function signatures would get cumbersome
      State = #state{checksums=Checksums,
                     org_id=OrgId,
                     bucket=Bucket,
                     token=Token},
  
      StateBeforeRemainder = lists:foldl(fun fetcher/2, State, Checksums),
      fetch_remainder(StateBeforeRemainder#state{count=Remainder}).
  
  fetcher(Checksum, #state{count=?MAX}=State) ->
      State1 = gather(State),
      fetcher(Checksum, State1#state{count=0, workers=[]});
  
  fetcher(Checksum, #state{count=Count,
                           token=Token,
                           org_id=OrgId,
                           bucket=Bucket,
                           workers=Workers}=State) ->
      Worker = spawn_worker(OrgId, Bucket, Checksum, Token),
      State#state{count=Count+1,
                  workers=[Worker|Workers]}.
  
  fetch_remainder(#state{}=State) ->
      #state{ok=Ok, errors=Errors, missing=Missing} = gather(State),
      Result = {{ok, lists:sort(Ok)},
                {missing, lists:sort(Missing)},
                {error, Errors}},
      Result.
  
  gather(#state{count=0}=State) ->
      State;
  gather(#state{token=Token, ok=Ok, missing=Missing, errors=Errors, count=Count, workers=Workers}=State) ->
      receive
          {Token, {Status, Checksum}, WorkerPid} ->
              ResultState0 = case Status of
                                 ok ->
                                     State#state{ok=[Checksum|Ok]};
                                 missing ->
                                     State#state{missing=[Checksum|Missing]};
                                 error ->
                                     State#state{errors=Errors + 1}
                             end,
              ResultState = ResultState0#state{count=Count-1,
                                               workers=lists:delete(WorkerPid, Workers)},
              gather(ResultState)
      after ?TIMEOUT ->
              %% Kill remaining workers; they're taking too long
              [ erlang:exit(W, kill) || W <- Workers ],
              State#state{workers=[], errors=Errors + length(Workers)}
      end.
  
  -spec spawn_worker(OrgId :: object_id(),
                     Bucket :: string(),
                     Checksum :: binary(),
                     Token :: reference()) -> Worker :: pid().
  spawn_worker(OrgId, Bucket, Checksum, Token) ->
      Master = self(), %% This is the same as the process that is doing the scatter / gather
      erlang:spawn(fun() ->
                           Result = check_file(OrgId, Bucket, Checksum),
                           Master ! {Token, Result, self()}
                   end).
  
  %% @doc Check to see if a checksum file is stored in S3 already by checking for file
  %% metadata.
  -spec check_file(OrgId :: object_id(),
                   Bucket :: string(),
                   Checksum :: binary()) -> {'error', binary()} |
                                            {'missing', binary()} |
                                            {'ok',binary()}.
  check_file(OrgId, Bucket, Checksum) ->
      Key = chef_s3:make_key(OrgId, Checksum),
      AwsConfig = chef_s3:get_config(),
      Result = try mini_s3:get_object_metadata(Bucket, Key, [], AwsConfig) of
                   _V ->
                       {ok, Checksum}
               catch
                   error:{aws_error, {http_error,404,_}} ->
                       {missing, Checksum};
                   _X:_Y->
                       {error, Checksum}
               end,
      Result.
  
negative.code: 
  -module(chef_s3_ops).
  
  -include("chef_types.hrl").
  
  -export([
           delete/2,
           fetch_md/2
          ]).
  
  %% @doc Delete each checksummed file in S3.  
  %%
  %% Returns a tuple of tagged tuples indicating which checksums were successfully deleted and
  %% which were not found, along with both the number of timeouts that occured, as well as the
  %% number of other errors occurred.
  %%
  %% Note that the return value information is all disjoint; that is, the number of found and
  %% missing checksums, plus the number of errors and timeouts, will be equal to the length of
  %% `Checksums'.
  -spec delete(OrgId :: object_id(),
               Checksums :: [binary()]) -> {{ok, [binary()]},
                                            {missing, [binary()]},
                                            {timeout, non_neg_integer()},
                                            {error, non_neg_integer()}}.
  delete(OrgId, Checksums) when is_list(Checksums),
                                is_binary(OrgId) ->
  
      {Ok, Missing, NumTimeouts, NumErrors} = parallelize(OrgId, Checksums, delete_file),
  
      {{ok, Ok},
       {missing, Missing},
       {timeout, NumTimeouts},
       {error, NumErrors}}.
  
  %% @doc Verify that each checksummed file is stored in S3 by checking its metadata.
  %%
  %% Returns a tuple of tagged tuples indicating which checksums were found and which were
  %% not, along with both the number of errors (including timeouts, as distinct from
  %% `delete/2') that occurred.
  %%
  %% Note that the return value information is all disjoint; that is, the number of found and
  %% missing checksums, plus the number of errors, will be equal to the length of `Checksums'.
  -spec fetch_md(OrgId :: object_id(),
                Checksums :: [binary()]) -> {{ok, [binary()]},
                                             {missing, [binary()]},
                                             {error, non_neg_integer()}}.
  fetch_md(OrgId, Checksums) when is_list(Checksums),
                                  is_binary(OrgId) ->
      
      {Ok, Missing, NumTimeouts, NumErrors} = parallelize(OrgId, Checksums, check_file),
      
      {{ok, Ok},
       {missing, Missing},
       %% Currently, for file checking purposes, a timeout is considered an error
       {error, NumErrors + NumTimeouts}}.
  
  %% HTTP Operation Functions
  %%
  %% These functions are responsible for actually issuing an HTTP request to S3 to either
  %% delete a file or to verify its existence.
  %%
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  
  %% @doc Delete an existing file (identified by `Checksum') in S3
  -spec delete_file(OrgId :: object_id(),
                    AwsConfig :: mini_s3:config(),
                    Bucket :: string(),
                    Checksum :: binary()) -> {'error', binary()} |
                                             {'missing', binary()} |
                                             {'ok', binary()}.
  delete_file(OrgId, AwsConfig, Bucket, Checksum) ->
      Key = chef_s3:make_key(OrgId, Checksum),
  
      try mini_s3:delete_object(Bucket, Key, AwsConfig) of
          %% Return value doesn't much matter here but for documentation's sake it looks like:
          %%
          %%  [{delete_marker, list_to_existing_atom(Marker)}, {version_id, Id}]
          %%
          %% The main take-away here is that the call to delete_object/3 was successful and
          %% didn't throw an error
          _Response ->
              {ok, Checksum}
      catch
          error:{aws_error, {http_error,404,_}} ->
              %% We got a 404.  Since this *may* be indicative of weirdness, we'll log it, but
              %% we don't need to crash or raise an exception.
              error_logger:warning_msg("Deletion of file (checksum: ~p) for org ~p from bucket ~p (key: ~p) failed because the file was not found~n",
                                       [Checksum, OrgId, Bucket, Key]),
              {missing, Checksum};
          ExceptionClass:Reason->
              %% Something unanticipated happened.  We should log the specific reason for
              %% later analysis, but as far as the overall deletion operation is concerned,
              %% this is "just an error", and we can continue along.
              error_logger:error_msg("Deletion of file (checksum: ~p) for org ~p from bucket ~p (key: ~p) raised exception ~p:~p~n",
                                     [Checksum, OrgId, Bucket, Key, ExceptionClass, Reason]),
              {error, Checksum}
      end.
  
  %% @doc See if the file represented by the given `Checksum' exists in S3.
  -spec check_file(OrgId :: object_id(),
                   AwsConfig :: mini_s3:config(),
                   Bucket :: string(),
                   Checksum :: binary()) -> {'error', binary()} |
                                            {'missing', binary()} |
                                            {'ok', binary()}.
  check_file(OrgId, AwsConfig, Bucket, Checksum) ->
      Key = chef_s3:make_key(OrgId, Checksum),
  
      Result = try mini_s3:get_object_metadata(Bucket, Key, [], AwsConfig) of
                   _Response ->
                       %% Actual return value doesn't matter, just that the call didn't throw
                       %% an error.
                       {ok, Checksum}
               catch
                   error:{aws_error, {http_error,404,_}} ->
                       %% The file wasn't found.  Not logging it because this is not
                       %% necessarily indicative of badness
                       {missing, Checksum};
                   ExceptionClass:Reason->
                       %% Something unanticipated happened.  We should log the specific reason
                       %% for later analysis, but as far as the overall checking operation is
                       %% concerned, this is "just an error", and we can continue along.
                       error_logger:error_msg("Checking presence of file (checksum: ~p) for org ~p from bucket ~p (key: ~p) raised exception ~p:~p~n",
                                              [Checksum, OrgId, Bucket, Key, ExceptionClass, Reason]),
                       {error, Checksum}
               end,
      Result.
  
  %% Private Utility Functions
  %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  
  %% @doc Specifies how many requests to S3 / Bookshelf are in-flight at a given time.
  -spec fanout() -> Size :: pos_integer().
  