Repository: aws-solutions-library-samples/osml-model-runner (language: python)

Issue:
interface for generic mr post processing
*Issue #, if available:*

*Description of changes:*
Updating model runner to use a more generic "MRPostProcessing" list to read in feature selection/fusion options.

Sample Post-processing with NMS:

```
[
    {
        "step": "FEATURE_DISTILLATION",
        "algorithm": {
            "algorithmType": "NMS",
            "iouThreshold": 0.75
        }
    }
]
```

Sample Post-processing with Soft NMS:

```
[
    {
        "step": "FEATURE_DISTILLATION",
        "algorithm": {
            "algorithmType": "SOFT_NMS",
            "iouThreshold": 0.75,
            "skipBoxThreshold": 0.0001,
            "sigma": 0.1
        }
    }
]
```



Added unit tests for new code, passed centerpoint, flood, and aircraft integ tests.  Also used cesium globe to successfully submit image requests for small and large image with aircraft model.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.



Discussion / hints:

Fix patch:
diff --git a/.coveragerc b/.coveragerc
index 7286fe16..5100d4b5 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -17,4 +17,4 @@ partial_branches =
 
 show_missing = True
 
-fail_under = 80
+fail_under = 88
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index e70e97d8..05b63310 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1,46 +1,40 @@
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.4.0
+    rev: v4.5.0
     hooks:
       - id: check-yaml
       - id: end-of-file-fixer
       - id: trailing-whitespace
 
   - repo: https://github.com/psf/black
-    rev: 23.1.0
+    rev: 23.12.1
     hooks:
       - id: black
         args: ["--line-length=125"]
         types: [file, python]
 
   - repo: https://github.com/PyCQA/isort
-    rev: 5.12.0
+    rev: 5.13.2
     hooks:
       - id: isort
         args: ["--line-length=125", "--profile=black"]
         types: [file, python]
 
   - repo: https://github.com/PyCQA/flake8
-    rev: 6.0.0
+    rev: 6.1.0
     hooks:
       - id: flake8
         args: ["--ignore=E203,W503,W605", "--max-line-length=125"]
         types: [file, python]
 
-  - repo: https://github.com/pre-commit/mirrors-autopep8
-    rev: v2.0.1
+  - repo: https://github.com/hhatto/autopep8
+    rev: v2.0.4
     hooks:
       - id: autopep8
-        args:
-          [
-            '--ignore=["E203","W503","W605"]',
-            "--max-line-length=125",
-            "--in-place"
-          ]
         types: [file, python]
 
   - repo: https://github.com/aws-samples/automated-security-helper
-    rev: '1.0.9-e-16May2023' # update with the latest tagged version in the repository
+    rev: '1.1.0-e-01Dec2023' # update with the latest tagged version in the repository
     hooks:
         - id: ash
           stages: [ manual ]
diff --git a/pyproject.toml b/pyproject.toml
index a8da5df0..a2e29df4 100755
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -19,4 +19,12 @@ testpaths = [
     "test"
 ]
 
+[tool.autopep8]
+max_line_length = 125
+ignore = "E203,W503,W605"
+in-place = true
+recursive = true
+aggressive = 3
+
+
 # Note Flake8 can't currently be configured in pyproject.toml. See .flake8 for the configuration settings.
diff --git a/setup.cfg b/setup.cfg
index 407c7065..90e00eb4 100755
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,6 +1,6 @@
 [metadata]
 name = osml-model-runner
-version = 1.2.0
+version = 1.3.0
 description = Application to run large scale imagery against AI/ML models
 long_description = file: README.md
 long_description_content_type = text/markdown
@@ -39,9 +39,9 @@ python_requires = >=3.8
 include_package_data = True
 
 install_requires =
-    osml-imagery-toolkit>=1.1.1
+    osml-imagery-toolkit>=1.2.0
     numpy>=1.23.0
-    shapely>=1.8.5,<2.0.0
+    shapely>=1.8.5
     aws-embedded-metrics==3.1.0
     boto3==1.28.1
     botocore==1.31.1
diff --git a/src/aws/osml/model_runner/api/image_request.py b/src/aws/osml/model_runner/api/image_request.py
index 6d865b8f..14e31e31 100755
--- a/src/aws/osml/model_runner/api/image_request.py
+++ b/src/aws/osml/model_runner/api/image_request.py
@@ -1,6 +1,7 @@
 #  Copyright 2023 Amazon.com, Inc. or its affiliates.
 
 import logging
+from json import dumps, loads
 from typing import Any, Dict, List, Optional
 
 import boto3
@@ -10,10 +11,14 @@
 
 from aws.osml.model_runner.app_config import BotoConfig
 from aws.osml.model_runner.common import (
-    FeatureSelectionOptions,
+    FeatureDistillationAlgorithm,
+    FeatureDistillationNMS,
     ImageCompression,
     ImageDimensions,
     ImageFormats,
+    MRPostProcessing,
+    MRPostprocessingStep,
+    deserialize_post_processing_list,
     get_credentials_for_assumed_role,
 )
 
@@ -42,6 +47,10 @@ def __init__(self, *initial_data: Dict[str, Any], **kwargs: Any):
                              attributes
         :param kwargs: Any = keyword arguments provided on the constructor to set specific attributes
         """
+        default_post_processing = [
+            MRPostProcessing(step=MRPostprocessingStep.FEATURE_DISTILLATION, algorithm=FeatureDistillationNMS())
+        ]
+
         self.job_id: str = ""
         self.job_arn: str = ""
         self.image_id: str = ""
@@ -57,7 +66,7 @@ def __init__(self, *initial_data: Dict[str, Any], **kwargs: Any):
         self.model_invocation_role: str = ""
         self.feature_properties: List[dict] = []
         self.roi: Optional[BaseGeometry] = None
-        self.feature_selection_options: FeatureSelectionOptions = FeatureSelectionOptions()
+        self.post_processing: List[MRPostProcessing] = default_post_processing
 
         for dictionary in initial_data:
             for key in dictionary:
@@ -122,6 +131,14 @@ def from_external_message(image_request: Dict[str, Any]):
             ]
         if image_request.get("featureProperties"):
             properties["feature_properties"] = image_request["featureProperties"]
+        if image_request.get("postProcessing"):
+            image_request["postProcessing"] = loads(
+                dumps(image_request["postProcessing"])
+                .replace("algorithmType", "algorithm_type")
+                .replace("iouThreshold", "iou_threshold")
+                .replace("skipBoxThreshold", "skip_box_threshold")
+            )
+            properties["post_processing"] = deserialize_post_processing_list(image_request.get("postProcessing"))
 
         return ImageRequest(properties)
 
@@ -133,9 +150,16 @@ def is_valid(self) -> bool:
                  False otherwise
         """
         if not shared_properties_are_valid(self):
+            logger.error("Invalid shared properties in ImageRequest")
             return False
 
         if not self.job_arn or not self.job_id or not self.outputs:
+            logger.error("Missing job arn, job id, or outputs properties in ImageRequest")
+            return False
+
+        num_feature_detection_options = len(self.get_feature_distillation_option())
+        if num_feature_detection_options > 1:
+            logger.error("{} feature distillation options in ImageRequest".format(num_feature_detection_options))
             return False
 
         return True
@@ -160,6 +184,18 @@ def get_shared_values(self) -> Dict[str, Any]:
             "tile_compression": self.tile_compression,
         }
 
+    def get_feature_distillation_option(self) -> List[FeatureDistillationAlgorithm]:
+        """
+        Parses the post-processing property and extracts the relevant feature distillation selection, if present
+        :return:
+        """
+        return [
+            op.algorithm
+            for op in self.post_processing
+            if op.step == MRPostprocessingStep.FEATURE_DISTILLATION
+            and isinstance(op.algorithm, FeatureDistillationAlgorithm)
+        ]
+
     @staticmethod
     def validate_image_path(image_url: str, assumed_role: str) -> bool:
         """
diff --git a/src/aws/osml/model_runner/app.py b/src/aws/osml/model_runner/app.py
index d08c1ca7..1e7370bf 100755
--- a/src/aws/osml/model_runner/app.py
+++ b/src/aws/osml/model_runner/app.py
@@ -14,8 +14,6 @@
 from aws_embedded_metrics.logger.metrics_logger import MetricsLogger
 from aws_embedded_metrics.metric_scope import metric_scope
 from aws_embedded_metrics.unit import Unit
-from dacite import Config as dacite_Config
-from dacite import from_dict
 from geojson import Feature
 from osgeo import gdal
 from osgeo.gdal import Dataset
@@ -33,16 +31,16 @@
 from .app_config import MetricLabels, ServiceConfig
 from .common import (
     EndpointUtils,
-    FeatureSelectionAlgorithm,
-    FeatureSelectionOptions,
+    FeatureDistillationDeserializer,
     GeojsonDetectionField,
     ImageDimensions,
     ImageRegion,
     ImageRequestStatus,
+    RegionRequestStatus,
     Timer,
     build_embedded_metrics_config,
-    feature_selection_options_factory,
     get_credentials_for_assumed_role,
+    mr_post_processing_options_factory,
 )
 from .database import EndpointStatisticsTable, FeatureTable, JobItem, JobTable, RegionRequestItem, RegionRequestTable
 from .exceptions import (
@@ -282,10 +280,12 @@ def process_image_request(self, image_request: ImageRequest, metrics: MetricsLog
                 image_url=image_request.image_url,
                 image_read_role=image_request.image_read_role,
                 feature_properties=dumps(image_request.feature_properties),
-                feature_selection_options=dumps(
-                    asdict(image_request.feature_selection_options, dict_factory=feature_selection_options_factory)
-                ),
             )
+            feature_distillation_option_list = image_request.get_feature_distillation_option()
+            if feature_distillation_option_list:
+                image_request_item.feature_distillation_option = dumps(
+                    asdict(feature_distillation_option_list[0], dict_factory=mr_post_processing_options_factory)
+                )
 
             # Start the image processing
             self.job_table.start_image_request(image_request_item)
@@ -482,14 +482,13 @@ def process_region_request(
             if in_progress >= max_regions:
                 if isinstance(metrics, MetricsLogger):
                     metrics.put_metric(MetricLabels.REGIONS_SELF_THROTTLED, 1, str(Unit.COUNT.value))
-                logger.info(f"Throttling region request. (Max: {max_regions} In-progress: {in_progress}")
+                logger.info("Throttling region request. (Max: {} In-progress: {}".format(max_regions, in_progress))
                 raise SelfThrottledRegionException
 
-        try:
-            if ServiceConfig.self_throttling:
-                # Increment the endpoint region counter
-                self.endpoint_statistics_table.increment_region_count(region_request.model_name)
+            # Increment the endpoint region counter
+            self.endpoint_statistics_table.increment_region_count(region_request.model_name)
 
+        try:
             with Timer(
                 task_str="Processing region {} {}".format(region_request.image_url, region_request.region_bounds),
                 metric_name=MetricLabels.REGION_LATENCY,
@@ -497,15 +496,10 @@ def process_region_request(
                 metrics_logger=metrics,
             ):
                 # Set up our threaded tile worker pool
-                tile_queue, tile_workers = setup_tile_workers(
-                    region_request,
-                    sensor_model,
-                    self.elevation_model,
-                    metrics,
-                )
+                tile_queue, tile_workers = setup_tile_workers(region_request, sensor_model, self.elevation_model, metrics)
 
                 # Process all our tiles
-                total_tile_count = process_tiles(
+                total_tile_count, tile_error_count = process_tiles(
                     region_request, tile_queue, tile_workers, raster_dataset, metrics, sensor_model
                 )
 
@@ -514,32 +508,32 @@ def process_region_request(
                 region_request_item = self.region_request_table.update_region_request(region_request_item)
 
             # Update the image request to complete this region
-            image_request_item = self.job_table.complete_region_request(region_request.image_id)
+            image_request_item = self.job_table.complete_region_request(region_request.image_id, bool(tile_error_count))
 
             # Update region request table if that region succeeded
-            region_request_item = self.region_request_table.complete_region_request(region_request_item)
+            region_request_item = self.region_request_table.complete_region_request(
+                region_request_item, self.calculate_region_status(total_tile_count, tile_error_count)
+            )
 
-            if ServiceConfig.self_throttling:
-                # Decrement the endpoint region counter
-                self.endpoint_statistics_table.decrement_region_count(region_request.model_name)
             # Write CloudWatch Metrics to the Logs
             if isinstance(metrics, MetricsLogger):
                 metrics.put_metric(MetricLabels.REGIONS_PROCESSED, 1, str(Unit.COUNT.value))
                 metrics.put_metric(MetricLabels.TILES_PROCESSED, total_tile_count, str(Unit.COUNT.value))
+
             # Return the updated item
             return image_request_item
-        except Exception as err:
-            logger.error("Failed to process image region: {}", err)
 
+        except Exception as err:
+            logger.error("Failed to process image region: {}".format(err))
             # update the table to take in that exception
             region_request_item.message = "Failed to process image region: {0}".format(err)
+            return self.fail_region_request(region_request_item, metrics)
 
+        finally:
+            # Decrement the endpoint region counter
             if ServiceConfig.self_throttling:
-                # Decrement the endpoint region counter
                 self.endpoint_statistics_table.decrement_region_count(region_request.model_name)
 
-            return self.fail_region_request(region_request_item, metrics)
-
     @staticmethod
     def load_image_request(
         image_request_item: JobItem,
@@ -673,7 +667,9 @@ def complete_image_request(
             # Sink the features into the right outputs
             is_write_succeeded = self.sync_features(image_request_item, features)
             if not is_write_succeeded:
-                raise AggregateOutputFeaturesException("Failed to write features to S3 or Kinesis! Please check the log...")
+                raise AggregateOutputFeaturesException(
+                    "Failed to write features to S3 or Kinesis! Please check the " "log..."
+                )
 
             if isinstance(metrics, MetricsLogger):
                 # Record model used for this image
@@ -716,8 +712,9 @@ def fail_region_request(
             metrics.put_metric(MetricLabels.PROCESSING_FAILURE, 1, str(Unit.COUNT.value))
             metrics.put_metric(MetricLabels.REGION_PROCESSING_ERROR, 1, str(Unit.COUNT.value))
         try:
-            region_request_item = self.region_request_table.complete_region_request(region_request_item, error=True)
-
+            region_request_item = self.region_request_table.complete_region_request(
+                region_request_item, RegionRequestStatus.FAILED
+            )
             return self.job_table.complete_region_request(region_request_item.image_id, error=True)
         except Exception as status_error:
             logger.error("Unable to update region status in job table")
@@ -763,17 +760,88 @@ def aggregate_features(
         if image_request_item.tile_size and image_request_item.tile_overlap:
             # Read all the features from DDB.
             features = feature_table.get_features(image_request_item.image_id)
+            logger.info(f"Total features aggregated: {len(features)}")
         else:
             raise AggregateFeaturesException("Tile size and overlap must be provided for feature aggregation")
         return features
 
+    @staticmethod
+    def identify_overlap(
+        feature: Feature, shape: Tuple[int, int], overlap: Tuple[int, int], origin: Tuple[int, int] = (0, 0)
+    ) -> Tuple[int, int, int, int]:
+        """
+        Generate a tuple that contains the min and max indexes of adjacent tiles or regions for a given feature. If
+        the min and max values for both x and y are the same then this feature does not touch an overlap region.
+
+        :param feature: the geojson Feature that must contain properties to identify its location in an image
+        :param shape: the width, height of the area in pixels
+        :param overlap: the x, y overlap between areas in pixels
+        :param origin: the x, y coordinate of the area in relation to the full image
+
+        :return: a tuple: minx, maxx, miny, maxy that identifies any overlap.
+        """
+        bbox = feature["properties"][GeojsonDetectionField.BOUNDS]
+
+        # If an offset origin was supplied adjust the bbox so the key is relative to the origin.
+        bbox = (bbox[0] - origin[0], bbox[1] - origin[1], bbox[2] - origin[0], bbox[3] - origin[1])
+
+        stride_x = shape[0] - overlap[0]
+        stride_y = shape[0] - overlap[1]
+
+        max_x_index = int(bbox[2] / stride_x)
+        max_y_index = int(bbox[3] / stride_y)
+
+        min_x_index = int(bbox[0] / stride_x)
+        min_y_index = int(bbox[1] / stride_y)
+        min_x_offset = int(bbox[0]) % stride_x
+        min_y_offset = int(bbox[1]) % stride_y
+
+        if min_x_offset < overlap[0] and min_x_index > 0:
+            min_x_index -= 1
+        if min_y_offset < overlap[1] and min_y_index > 0:
+            min_y_index -= 1
+
+        return min_x_index, max_x_index, min_y_index, max_y_index
+
+    @staticmethod
+    def group_features_by_overlap(
+        features: List[Feature], shape: Tuple[int, int], overlap: Tuple[int, int], origin: Tuple[int, int] = (0, 0)
+    ) -> Dict[Tuple[int, int, int, int], List[Feature]]:
+        """
+        Group all the feature items by tile id
+
+        :param features: List[FeatureItem] = the list of feature items
+        :param shape: the width, height of the area in pixels
+        :param overlap: the x, y overlap between areas in pixels
+        :param origin: the x, y coordinate of the area in relation to the full image
+
+        :return: a mapping of overlap id to a list of features that intersect that overlap region
+        """
+        grouped_features: Dict[Tuple[int, int, int, int], List[Feature]] = {}
+        for feature in features:
+            overlap_key = ModelRunner.identify_overlap(feature, shape, overlap, origin)
+            grouped_features.setdefault(overlap_key, []).append(feature)
+        return grouped_features
+
     @staticmethod
     @metric_scope
     def select_features(
         image_request_item: JobItem, features: List[Feature], metrics: MetricsLogger = None
     ) -> List[Feature]:
         """
-        Selects the desired features using the options in the JobItem (NMS, SOFT_NMS, etc.)
+        Selects the desired features using the options in the JobItem (NMS, SOFT_NMS, etc.).
+        This code applies a feature selector only to the features that came from regions of the image
+        that were processed multiple times. First features are grouped based on the region they were
+        processed in. Any features found in the overlap area between regions are run through the
+        FeatureSelector. If they were not part of an overlap area between regions, they will be grouped
+        based on tile boundaries. Any features that fall into the overlap of adjacent tiles are filtered
+        by the FeatureSelector. All other features should not be duplicates; they are added to the result
+        without additional filtering.
+
+        Computationally, this implements two critical factors that lower the overall processing time for the
+        O(N^2) selection algorithms. First, it will filter out the majority of features that couldn't possibly
+        have duplicates generated by our tiled image processing; Second, it runs the selection algorithms
+        incrementally on much smaller groups of features.
 
         :param image_request_item: JobItem = the image request
         :param features: List[Feature] = the list of geojson features to process
@@ -788,13 +856,46 @@ def select_features(
             logger=logger,
             metrics_logger=metrics,
         ):
-            feature_selection_options = from_dict(
-                data_class=FeatureSelectionOptions,
-                data=json.loads(image_request_item.feature_selection_options),
-                config=dacite_Config(cast=[FeatureSelectionAlgorithm]),
+            feature_distillation_option_dict = json.loads(image_request_item.feature_distillation_option)
+            feature_distillation_option = FeatureDistillationDeserializer().deserialize(feature_distillation_option_dict)
+            feature_selector = FeatureSelector(feature_distillation_option)
+
+            region_size = ast.literal_eval(ServiceConfig.region_size)
+            tile_size = ast.literal_eval(image_request_item.tile_size)
+            overlap = ast.literal_eval(image_request_item.tile_overlap)
+
+            logger.debug("FeatureSelection: Starting overlap-aware deduplication of features.")
+            total_skipped = 0
+            deduped_features = []
+            features_grouped_by_region = ModelRunner.group_features_by_overlap(features, region_size, overlap)
+            for region_key, region_features in features_grouped_by_region.items():
+                region_stride = (region_size[0] - overlap[0], region_size[1] - overlap[1])
+                region_origin = (region_stride[0] * region_key[0], region_stride[1] * region_key[1])
+
+                if region_key[0] != region_key[1] or region_key[2] != region_key[3]:
+                    # The Group contains contributions from multiple regions, run selection on the entire group
+                    deduped_features.extend(feature_selector.select_features(region_features))
+                else:
+                    # Not an overlap between regions group these features using tile size to identify overlaps
+                    features_grouped_by_tile = ModelRunner.group_features_by_overlap(
+                        region_features, tile_size, overlap, region_origin
+                    )
+
+                    for tile_key, tile_features in features_grouped_by_tile.items():
+                        if tile_key[0] != tile_key[1] or tile_key[2] != tile_key[3]:
+                            # Group contains contributions from multiple tiles, run selection
+                            deduped_features.extend(feature_selector.select_features(tile_features))
+                        else:
+                            # No overlap between tiles, features can be added directly to the result
+                            total_skipped += len(tile_features)
+                            deduped_features.extend(tile_features)
+
+            logger.debug(
+                f"FeatureSelection: Skipped processing of {total_skipped} of {len(features)} features. "
+                "They were not inside an overlap region."
             )
-            feature_selector = FeatureSelector(feature_selection_options)
-            return feature_selector.select_features(features)
+
+            return deduped_features
 
     @staticmethod
     def sync_features(image_request_item: JobItem, features: List[Feature]) -> bool:
@@ -894,7 +995,7 @@ def get_inference_metadata_property(image_request_item: JobItem, inference_time:
                 "filePath": image_request_item.image_url,
                 "receiveTime": receive_time,
                 "inferenceTime": inference_time,
-                "tileOverlapFeatureSelection": image_request_item.feature_selection_options,
+                "tileOverlapFeatureSelection": image_request_item.feature_distillation_option,
             }
         }
         return inference_metadata_property
@@ -904,9 +1005,9 @@ def get_extents(ds: gdal.Dataset) -> dict[str, Any]:
         """
         Returns a list of driver extensions
 
-        :param ds: the gdal dateaset
+        :param ds: the gdal dataset
 
-        :return: List[number] = the extents of the image
+        :return: List[number] = the extent of the image
         """
         geo_transform = ds.GetGeoTransform()
         minx = geo_transform[0]
@@ -916,3 +1017,18 @@ def get_extents(ds: gdal.Dataset) -> dict[str, Any]:
 
         extents = {"north": maxy, "south": miny, "east": maxx, "west": minx}
         return extents
+
+    @staticmethod
+    def calculate_region_status(total_tile_count: int, tile_error_count: int) -> RegionRequestStatus:
+        """
+        Calculate the processing status of a region upon completion
+        :param total_tile_count: number of tiles that were processed
+        :param tile_error_count: number of tiles with errors
+        :return: RegionRequestStatus
+        """
+        region_status = RegionRequestStatus.SUCCESS
+        if total_tile_count == tile_error_count:
+            region_status = RegionRequestStatus.FAILED
+        if 0 < tile_error_count < total_tile_count:
+            region_status = RegionRequestStatus.PARTIAL
+        return region_status
diff --git a/src/aws/osml/model_runner/app_config.py b/src/aws/osml/model_runner/app_config.py
index 3b641693..45621958 100755
--- a/src/aws/osml/model_runner/app_config.py
+++ b/src/aws/osml/model_runner/app_config.py
@@ -77,9 +77,11 @@ class BotoConfig:
     sagemaker: (Config) the sagemaker specific boto client configuration
     """
 
-    # required env configuration
     default: Config = Config(region_name=ServiceConfig.aws_region, retries={"max_attempts": 15, "mode": "standard"})
     sagemaker: Config = Config(region_name=ServiceConfig.aws_region, retries={"max_attempts": 30, "mode": "adaptive"})
+    ddb: Config = Config(
+        region_name=ServiceConfig.aws_region, retries={"max_attempts": 3, "mode": "standard"}, max_pool_connections=50
+    )
 
 
 class MetricLabels(str, Enum):
diff --git a/src/aws/osml/model_runner/common/__init__.py b/src/aws/osml/model_runner/common/__init__.py
index 29c1e466..f46b0c69 100755
--- a/src/aws/osml/model_runner/common/__init__.py
+++ b/src/aws/osml/model_runner/common/__init__.py
@@ -8,9 +8,21 @@
 from .credentials_utils import get_credentials_for_assumed_role
 from .endpoint_utils import EndpointUtils
 from .exceptions import InvalidAssumedRoleException, InvalidClassificationException
-from .feature_selection_algorithm import FeatureSelectionAlgorithm
-from .feature_selection_options import FeatureSelectionOptions, feature_selection_options_factory
 from .metrics_utils import build_embedded_metrics_config
+from .mr_post_processing import (
+    FeatureDistillationAlgorithm,
+    FeatureDistillationAlgorithmType,
+    FeatureDistillationDeserializer,
+    FeatureDistillationNMS,
+    FeatureDistillationSoftNMS,
+    MRPostProcessing,
+    MRPostProcessingAlgorithm,
+    MRPostProcessingAlgorithmType,
+    MRPostProcessingDeserializer,
+    MRPostprocessingStep,
+    deserialize_post_processing_list,
+    mr_post_processing_options_factory,
+)
 from .security_classification import (
     Classification,
     ClassificationLevel,
@@ -28,4 +40,5 @@
     ImageFormats,
     ImageRegion,
     ImageRequestStatus,
+    RegionRequestStatus,
 )
diff --git a/src/aws/osml/model_runner/common/feature_selection_algorithm.py b/src/aws/osml/model_runner/common/feature_selection_algorithm.py
deleted file mode 100755
index 64be2aea..00000000
--- a/src/aws/osml/model_runner/common/feature_selection_algorithm.py
+++ /dev/null
@@ -1,15 +0,0 @@
-#  Copyright 2023 Amazon.com, Inc. or its affiliates.
-
-from enum import auto
-
-from aws.osml.model_runner.common import AutoStringEnum
-
-
-class FeatureSelectionAlgorithm(str, AutoStringEnum):
-    """
-    Enum defining the entity selection algorithm used in combining bounding boxes.
-    """
-
-    NONE = auto()
-    NMS = auto()
-    SOFT_NMS = auto()  # gaussian Soft-NMS (as opposed to linear)
diff --git a/src/aws/osml/model_runner/common/feature_selection_options.py b/src/aws/osml/model_runner/common/feature_selection_options.py
deleted file mode 100755
index 742972fd..00000000
--- a/src/aws/osml/model_runner/common/feature_selection_options.py
+++ /dev/null
@@ -1,32 +0,0 @@
-#  Copyright 2023 Amazon.com, Inc. or its affiliates.
-
-from dataclasses import dataclass
-from typing import Dict, List
-
-from aws.osml.model_runner.common.feature_selection_algorithm import FeatureSelectionAlgorithm
-
-
-@dataclass
-class FeatureSelectionOptions:
-    """
-    :property iou_threshold: float = intersection over union threshold
-                                    - if greater than this value boxes are considered the same
-    :property algorithm: FeatureSelectionAlgorithm = algorithm to use to combine object detections
-    :property skip_box_threshold: float = boxes with a confidence below this threshold value are skipped
-    :property sigma: float = value - only applies to Soft NMS
-    """
-
-    algorithm: FeatureSelectionAlgorithm = FeatureSelectionAlgorithm.NMS
-    iou_threshold: float = 0.75
-    skip_box_threshold: float = 0.0001
-    sigma: float = 0.1
-
-
-def feature_selection_options_factory(properties_list: List) -> Dict:
-    """
-    Factory to facilitate converting the dataclass to a dict.
-    Used with asdict(my_feature_selection_object, dict_factory=feature_selection_options_factory)
-    :param properties_list: list of properties
-    :return: dictionary representation of dataclass
-    """
-    return {k: (v.name if isinstance(v, FeatureSelectionAlgorithm) else v) for k, v in properties_list}
diff --git a/src/aws/osml/model_runner/common/mr_post_processing.py b/src/aws/osml/model_runner/common/mr_post_processing.py
new file mode 100755
index 00000000..87cc2366
--- /dev/null
+++ b/src/aws/osml/model_runner/common/mr_post_processing.py
@@ -0,0 +1,193 @@
+#  Copyright 2023 Amazon.com, Inc. or its affiliates.
+
+from abc import ABC, ABCMeta, abstractmethod
+from dataclasses import dataclass, field
+from enum import Enum, EnumMeta, auto
+from typing import Dict, List
+
+from .auto_string_enum import AutoStringEnum
+
+
+class ABCEnumMeta(ABCMeta, EnumMeta):
+    """
+    A metaclass combining ABCMeta and EnumMeta, used as a base for creating
+    enums for MRPostProcessing algorithm types. It ensures that the derived enums
+    are both Enums and ABCs, enabling type checks and abstract method enforcement.
+    """
+
+    def __new__(mcls, *args, **kw):
+        cls = super().__new__(mcls, *args, **kw)
+        if issubclass(cls, Enum) and getattr(cls, "__abstractmethods__", None):
+            raise TypeError("...")
+        return cls
+
+    def __call__(cls, *args, **kw):
+        if getattr(cls, "__abstractmethods__", None):
+            raise TypeError(
+                f"Can't instantiate abstract class {cls.__name__} " f"with frozen methods {set(cls.__abstractmethods__)}"
+            )
+        return super().__call__(*args, **kw)
+
+
+class MRPostProcessingAlgorithmType(AutoStringEnum, metaclass=ABCEnumMeta):
+    pass
+
+
+class FeatureDistillationAlgorithmType(str, MRPostProcessingAlgorithmType):
+    """
+    Enum for defining different feature distillation algorithms used in post-processing.
+    Each member represents a specific algorithm for entity selection or fusion.
+    NMS: Non-maximum Suppression
+    SOFT_NMS: Variant of NMS (https://arxiv.org/abs/1704.04503). This implementation is gaussian Soft-NMS,
+                as opposed to linear.
+    NMW: Non-maximum weighted
+    WBF: Weighted boxes fusion
+    """
+
+    NMS = auto()
+    SOFT_NMS = auto()
+
+
+class MRPostprocessingStep(str, AutoStringEnum):
+    """
+    Enum defining available steps for MR post-processing.
+    """
+
+    FEATURE_DISTILLATION = auto()
+
+
+@dataclass(frozen=True)
+class MRPostProcessingAlgorithm(ABC):
+    """
+    Abstract base class representing a generic MR post-processing algorithm.
+    :param algorithm_type: The type of the algorithm, represented by a specific enum.
+    """
+
+    algorithm_type: MRPostProcessingAlgorithmType
+
+
+@dataclass(frozen=True)
+class FeatureDistillationAlgorithm(MRPostProcessingAlgorithm):
+    """
+    Generic FeatureDistillationAlgorithm class that is designed to be extended for each specific algorithm.
+    :property algorithm_type: FeatureSelectionAlgorithmType = algorithm to use to combine object detections
+    :property iou_threshold: float = intersection over union threshold
+                                    - if greater than this value boxes are considered the same
+    """
+
+    algorithm_type: FeatureDistillationAlgorithmType
+    iou_threshold: float
+
+
+@dataclass(frozen=True)
+class FeatureDistillationNMS(FeatureDistillationAlgorithm):
+    """
+    :property algorithm_type: FeatureSelectionAlgorithmType = algorithm to use to combine object detections
+    :property iou_threshold: float = intersection over union threshold
+                                    - if greater than this value boxes are considered the same
+    """
+
+    algorithm_type: FeatureDistillationAlgorithmType = field(default=FeatureDistillationAlgorithmType.NMS)
+    iou_threshold: float = field(default=0.75)
+
+
+@dataclass(frozen=True)
+class FeatureDistillationSoftNMS(FeatureDistillationAlgorithm):
+    """
+    :property algorithm_type: FeatureSelectionAlgorithmType = algorithm to use to combine object detections
+    :property iou_threshold: float = intersection over union threshold
+                                    - if greater than this value boxes are considered the same
+    :property skip_box_threshold: float = boxes with a confidence below this threshold value are skipped
+    :property sigma: float = value - only applies to Soft NMS
+    """
+
+    algorithm_type: FeatureDistillationAlgorithmType = field(default=FeatureDistillationAlgorithmType.SOFT_NMS)
+    iou_threshold: float = field(default=0.75)
+    skip_box_threshold: float = field(default=0.0001)
+    sigma: float = field(default=0.1)
+
+
+@dataclass(frozen=True)
+class MRPostProcessing:
+    """
+    Represents an operation for MR post-processing.
+    :param step: The post-processing step (MRPostprocessingStep).
+    :param algorithm: The algorithm used in this step (MRPostProcessingAlgorithm).
+    """
+
+    step: MRPostprocessingStep
+    algorithm: MRPostProcessingAlgorithm
+
+
+class PostProcessingDeserializer(ABC):
+    """
+    Abstract post-processing deserializer that defines a deserialize method.
+    """
+
+    @staticmethod
+    @abstractmethod
+    def deserialize(post_processing_algorithm: dict) -> MRPostProcessingAlgorithm:
+        pass
+
+
+class FeatureDistillationDeserializer(PostProcessingDeserializer):
+    """
+    Provides static methods for deserializing feature distillation algorithms from a dictionary.
+    """
+
+    @staticmethod
+    def deserialize(post_processing_algorithm: dict) -> FeatureDistillationAlgorithm:
+        if "algorithm_type" in post_processing_algorithm:
+            post_processing_algorithm["algorithm_type"] = FeatureDistillationAlgorithmType(
+                post_processing_algorithm["algorithm_type"]
+            )
+            if post_processing_algorithm["algorithm_type"] == FeatureDistillationAlgorithmType.NMS:
+                return FeatureDistillationNMS(**post_processing_algorithm)
+            elif post_processing_algorithm["algorithm_type"] == FeatureDistillationAlgorithmType.SOFT_NMS:
+                return FeatureDistillationSoftNMS(**post_processing_algorithm)
+
+        raise ValueError(
+            f"Failed to deserialize. {post_processing_algorithm} is not a valid feature distillation algorithm object."
+        )
+
+
+class MRPostProcessingDeserializer:
+    """
+    Provides static method for obtaining the appropriate deserializer based on the post-processing step.
+    """
+
+    @staticmethod
+    def get_deserializer(step: MRPostprocessingStep) -> PostProcessingDeserializer:
+        if step == MRPostprocessingStep.FEATURE_DISTILLATION:
+            return FeatureDistillationDeserializer()
+        else:
+            raise ValueError("Failed to get deserializer. MR post-processing step not valid.")
+
+
+def deserialize_post_processing_list(mr_processing_list: List) -> List[MRPostProcessing]:
+    """
+    Deserializes a list of MR post-processing steps and algorithms from a list of dictionaries.
+    :param mr_processing_list: A list of dictionaries representing MR post-processing objects.
+    :return: A list of MRPostProcessing objects.
+    """
+    try:
+        post_processing_list: List[MRPostProcessing] = []
+        for mr_processing_dict in mr_processing_list:
+            step = MRPostprocessingStep(mr_processing_dict["step"])
+            algorithm_dict = mr_processing_dict["algorithm"]
+            deserializer = MRPostProcessingDeserializer().get_deserializer(step)
+            post_processing_list.append(MRPostProcessing(step=step, algorithm=deserializer.deserialize(algorithm_dict)))
+        return post_processing_list
+
+    except Exception as err:
+        raise err
+
+
+def mr_post_processing_options_factory(properties_list: List) -> Dict:
+    """
+    Factory function for converting dataclass instances to dictionaries.
+    Specifically used with MR post-processing related dataclasses.
+    :param properties_list: A list of properties (key-value pairs) of a dataclass.
+    :return: A dictionary representation of the dataclass.
+    """
+    return {k: (v.name if isinstance(v, MRPostProcessingAlgorithmType) else v) for k, v in properties_list}
diff --git a/src/aws/osml/model_runner/common/typing.py b/src/aws/osml/model_runner/common/typing.py
index 2453292c..906a06e7 100755
--- a/src/aws/osml/model_runner/common/typing.py
+++ b/src/aws/osml/model_runner/common/typing.py
@@ -47,6 +47,18 @@ class ImageRequestStatus(str, AutoStringEnum):
     FAILED = auto()
 
 
+class RegionRequestStatus(str, AutoStringEnum):
+    """
+    Enumeration defining status for region
+    """
+
+    STARTING = auto()
+    PARTIAL = auto()
+    IN_PROGRESS = auto()
+    SUCCESS = auto()
+    FAILED = auto()
+
+
 class GeojsonDetectionField(str, Enum):
     """
     Enumeration defining the model geojson field to index depending on the shape
diff --git a/src/aws/osml/model_runner/database/ddb_helper.py b/src/aws/osml/model_runner/database/ddb_helper.py
index a0cab1bc..eafee45d 100755
--- a/src/aws/osml/model_runner/database/ddb_helper.py
+++ b/src/aws/osml/model_runner/database/ddb_helper.py
@@ -1,15 +1,17 @@
 #  Copyright 2023 Amazon.com, Inc. or its affiliates.
 
 import logging
+import random
+import time
 from dataclasses import asdict, dataclass, field, fields
 from typing import Any, Dict, List, Optional, Tuple
 
 import boto3
-from boto3 import dynamodb
+from boto3.dynamodb.conditions import Key
 
 from aws.osml.model_runner.app_config import BotoConfig
 
-from .exceptions import DDBUpdateException
+from .exceptions import DDBBatchWriteException, DDBUpdateException
 
 logger = logging.getLogger(__name__)
 
@@ -63,9 +65,9 @@ def __get_fields():
 
 class DDBHelper:
     """
-    DDBHelper is a class meant to help OSML with accessing and interacting with DynamoDB tables. Generally this class
-    should be inherited by downstream specific table classes to build on top of such as the FeatureTable and JobTable
-    classes.
+    DDBHelper is a class meant to help OSML with accessing and interacting with DynamoDB tables.
+    Generally, this class should be inherited by downstream specific table classes to build on top
+    of such as the FeatureTable and JobTable classes.
 
     :param table_name: str = the name of the table to interact with
 
@@ -74,8 +76,9 @@ class DDBHelper:
 
     def __init__(self, table_name: str) -> None:
         # build a table resource to use for accessing data
-        self.table = boto3.resource("dynamodb", config=BotoConfig.default).Table(table_name)
         self.table_name = table_name
+        self.client = boto3.resource("dynamodb", config=BotoConfig.ddb)
+        self.table = self.client.Table(table_name)
 
     def get_ddb_item(self, ddb_item: DDBItem) -> Dict[str, Any]:
         """
@@ -89,11 +92,10 @@ def get_ddb_item(self, ddb_item: DDBItem) -> Dict[str, Any]:
 
     def put_ddb_item(self, ddb_item: DDBItem, condition_expression: str = None) -> Dict[str, Any]:
         """
-        Put a DynamoDB item into the table
+        Put a DynamoDB item into the table with a jitter-delayed retry logic for unprocessed items.
 
         :param ddb_item: DDBItem = item that we want to put (required)
-        :param condition_expression: str = condition that must be satisfied in order for a
-                                            conditional PutItem operation to succeed.
+        :param condition_expression: str = Condition that must be satisfied in order for a PutItem operation to succeed.
 
         :return: Dict[str, Any] = item from the put_item response
         """
@@ -104,6 +106,58 @@ def put_ddb_item(self, ddb_item: DDBItem, condition_expression: str = None) -> D
 
         return response
 
+    def batch_write_items(self, ddb_items: List[DDBItem], max_retries: int = 500, max_delay: float = 8) -> None:
+        """
+        Write multiple DynamoDB items in a batch with a jitter-delayed retry logic for unprocessed items.
+        :param ddb_items: List[DDBItem] = List of items that we want to write.
+        :param max_retries: int = Maximum number of retries for unprocessed items.
+        :param max_delay: Maximum delay in seconds between retries.
+        :return: None
+        """
+
+        def _batch_write(items: Dict[str, Any], retries: int = 0, initial_delay: float = 0.125):
+            # Calculate smart jitter backoff for retries
+            delay = random.uniform(0, min(max_delay, initial_delay * 2**retries))
+            try:
+                response = self.client.batch_write_item(RequestItems=items)
+                unprocessed_items = response.get("UnprocessedItems", {})
+                # If we failed to process some items, try again
+                if unprocessed_items and retries < max_retries:
+                    time.sleep(delay)
+                    _batch_write(unprocessed_items, retries + 1)
+                # If we still have unprocessed items after the retry limit
+                elif unprocessed_items:
+                    raise DDBBatchWriteException(f"Failed to process items: {unprocessed_items}.")
+
+                logger.debug("Successfully batch wrote items to table.")
+            except Exception as err:
+                # If we failed to call the write try again
+                if retries < max_retries:
+                    time.sleep(delay)
+                    _batch_write(items, retries + 1)
+                else:
+                    logger.error(err)
+                    raise
+
+        # Max batch size DDB (DynamoDB) supports is 25
+        batch_size = 25
+
+        # Iterate over ddb_items in chunks of batch_size
+        for i in range(0, len(ddb_items), batch_size):
+            batch = ddb_items[i : i + batch_size]
+            request_items = {self.table_name: []}
+
+            # Prepare the batch request for DynamoDB
+            for item in batch:
+                # Ensure that the item has a method `to_put()` that formats it for DynamoDB
+                put_request = {"PutRequest": {"Item": item.to_put()}}
+                request_items[self.table_name].append(put_request)
+
+            # Send the batch write request
+            _batch_write(request_items)
+
+        return
+
     def delete_ddb_item(self, ddb_item: DDBItem) -> Dict[str, Any]:
         """
         Delete a DynamoDB item from the table
@@ -117,7 +171,7 @@ def delete_ddb_item(self, ddb_item: DDBItem) -> Dict[str, Any]:
     def update_ddb_item(self, ddb_item: DDBItem, update_exp: str = None, update_attr: Dict = None) -> Dict[str, Any]:
         """
         Update the DynamoDB item based on the contents of an input dictionary. If the user doesn't
-        provide an update expression and attributes one will be generated from the body.
+        provide an update expression and attributes, one will be generated from the body.
 
         :param ddb_item: DDBItem = item that we want to update (required)
         :param update_exp: Optional[str] = the update expression to use for the update
@@ -131,7 +185,7 @@ def update_ddb_item(self, ddb_item: DDBItem, update_exp: str = None, update_attr
             update_item = ddb_item.to_update()
             update_exp, update_attr = self.get_update_params(update_item, ddb_item)
 
-        # if we still don't have an update expression then we'll just
+        # if we still don't have an update expression, then we'll just
         if update_exp and update_attr:
             response = self.table.update_item(
                 Key=self.get_keys(ddb_item=ddb_item),
@@ -156,7 +210,7 @@ def query_items(self, ddb_item: DDBItem) -> List[Dict[str, Any]]:
         all_items_retrieved = False
         response = self.table.query(
             ConsistentRead=True,
-            KeyConditionExpression=dynamodb.conditions.Key(ddb_item.ddb_key.hash_key).eq(ddb_item.ddb_key.hash_value),
+            KeyConditionExpression=Key(ddb_item.ddb_key.hash_key).eq(ddb_item.ddb_key.hash_value),
         )
 
         # Grab all the items from the table
@@ -167,9 +221,7 @@ def query_items(self, ddb_item: DDBItem) -> List[Dict[str, Any]]:
             if "LastEvaluatedKey" in response:
                 response = self.table.query(
                     ConsistentRead=True,
-                    KeyConditionExpression=dynamodb.conditions.Key(ddb_item.ddb_key.hash_key).eq(
-                        ddb_item.ddb_key.hash_value
-                    ),
+                    KeyConditionExpression=Key(ddb_item.ddb_key.hash_key).eq(ddb_item.ddb_key.hash_value),
                     ExclusiveStartKey=response["LastEvaluatedKey"],
                 )
             else:
@@ -205,7 +257,7 @@ def get_keys(ddb_item: DDBItem) -> Dict[str, Any]:
 
         :param ddb_item: DDBItem = the hash key we want to query the table for
 
-        :return Dict[str, Any] = Holding either Hash Key or both Keys (Hash and Range)
+        return Dict[str, Any] = Holding either Hash Key or both Keys (Hash and Range)
         """
         if ddb_item.ddb_key.range_key is None:
             return {
diff --git a/src/aws/osml/model_runner/database/exceptions.py b/src/aws/osml/model_runner/database/exceptions.py
index 88dece95..550c9a41 100755
--- a/src/aws/osml/model_runner/database/exceptions.py
+++ b/src/aws/osml/model_runner/database/exceptions.py
@@ -10,6 +10,10 @@ class DDBUpdateException(Exception):
     pass
 
 
+class DDBBatchWriteException(Exception):
+    pass
+
+
 class EndImageException(Exception):
     pass
 
diff --git a/src/aws/osml/model_runner/database/feature_table.py b/src/aws/osml/model_runner/database/feature_table.py
index 514901b9..79522587 100755
--- a/src/aws/osml/model_runner/database/feature_table.py
+++ b/src/aws/osml/model_runner/database/feature_table.py
@@ -1,7 +1,9 @@
 #  Copyright 2023 Amazon.com, Inc. or its affiliates.
 
 import logging
+import random
 import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
 from secrets import token_hex
 from typing import Dict, List, Optional
@@ -11,6 +13,7 @@
 from aws_embedded_metrics.logger.metrics_logger import MetricsLogger
 from aws_embedded_metrics.metric_scope import metric_scope
 from aws_embedded_metrics.unit import Unit
+from botocore.parsers import PROTOCOL_PARSERS
 from dacite import from_dict
 from geojson import Feature
 
@@ -22,6 +25,24 @@
 
 logger = logging.getLogger(__name__)
 
+# The logic below removes of some redundant type checking in the DDB client
+# https://maori.geek.nz/make-pythons-dynamodb-client-faster-with-this-one-simple-trick-2eb2888269ce
+parser = PROTOCOL_PARSERS.get("json")
+
+
+def handle_json_body(self, raw_body, shape) -> Dict[str, any]:
+    """
+
+    :param self: Boto core class object.
+    :param raw_body: Raw JSON body content to be type checked.
+    :param shape: The type enforced shape to check against (none).
+    :return: The parsed body as a JSON object.
+    """
+    return self._parse_body_as_json(raw_body)
+
+
+setattr(parser, "_handle_json_body", handle_json_body)
+
 
 @dataclass
 class FeatureItem(DDBItem):
@@ -55,32 +76,33 @@ def __init__(self, table_name: str, tile_size: ImageDimensions, overlap: ImageDi
         super().__init__(table_name)
         self.tile_size = tile_size
         self.overlap = overlap
+        self.hash_salt = 50
 
     @metric_scope
     def add_features(self, features: List[Feature], metrics: MetricsLogger = None):
         """
         Group all the features together and add/update an item in the DDB
 
-        :param features: List[Feature] = the list of features
-        :param metrics: MetricsLogger = the metrics logger to use to report metrics.
-
+        :param features: The list of features to add to the DDB table.
+        :param metrics: Metrics logger to use to report metrics.
         :return: None
         """
         if isinstance(metrics, MetricsLogger):
             metrics.set_dimensions()
         start_time_millisec = int(time.time() * 1000)
         # These records are temporary and will expire 24 hours after creation. Jobs should take
-        # minutes to run so this time should be conservative enough to let a team debug an urgent
+        # minutes to run, so this time should be conservative enough to let a team debug an urgent
         # issue without leaving a ton of state leftover in the system.
-        expire_time_epoch_sec = Decimal(int(start_time_millisec / 1000) + (24 * 60 * 60))
+        expire_time_epoch_sec = Decimal(int(start_time_millisec / 1000) + (2 * 60 * 60))
         with Timer(
             task_str="Add image features",
             metric_name=MetricLabels.FEATURE_STORE_LATENCY,
             logger=logger,
             metrics_logger=metrics,
         ):
-            for key, grouped_features in self.group_features_by_key(features).items():
-                try:
+            try:
+                items = []
+                for key, grouped_features in self.group_features_by_key(features).items():
                     image_id, tile_id = key.split("-region-", 1)
 
                     logger.debug(
@@ -95,10 +117,10 @@ def add_features(self, features: List[Feature], metrics: MetricsLogger = None):
                         encoded_feature = geojson.dumps(feature)
                         total_encoded_length += len(encoded_feature)
                         encoded_features.append(encoded_feature)
-                        # Once we exceed the 200K byte limit on our features write them to DDB. We are
+                        # Once we exceed the 200K byte limit on our features, write them to DDB. We are
                         # batching at this size because a single row in DDB only allows for 400K. We also
                         # need to make sure we are processing the last item no matter what the size is.
-                        if total_encoded_length > int(ServiceConfig.ddb_max_item_size) or feature_count >= len(
+                        if total_encoded_length > int(ServiceConfig.ddb_max_item_size) or feature_count == len(
                             grouped_features
                         ):
                             logger.debug(
@@ -107,9 +129,9 @@ def add_features(self, features: List[Feature], metrics: MetricsLogger = None):
                             )
 
                             # Build up a feature item and put it in the table
-                            result = self.put_ddb_item(
+                            items.append(
                                 FeatureItem(
-                                    hash_key=image_id,
+                                    hash_key=image_id + "-" + str(random.randint(1, self.hash_salt)),
                                     range_key=token_hex(16),
                                     tile_id=tile_id,
                                     features=encoded_features,
@@ -121,72 +143,71 @@ def add_features(self, features: List[Feature], metrics: MetricsLogger = None):
                             total_encoded_length = 0
                             encoded_features = []
 
-                            # Check that we got a success response
-                            status_code = result.get("ResponseMetadata", {}).get("HTTPStatusCode")
-                            if status_code != 200:
-                                if isinstance(metrics, MetricsLogger):
-                                    metrics.put_metric(MetricLabels.FEATURE_UPDATE, 1, str(Unit.COUNT.value))
-                                    metrics.put_metric(MetricLabels.FEATURE_ERROR, 1, str(Unit.COUNT.value))
-                                logger.error("Unable to update feature table - HTTP Status Code: {}".format(status_code))
-                except Exception as err:
-                    if isinstance(metrics, MetricsLogger):
-                        metrics.put_metric(MetricLabels.FEATURE_UPDATE_EXCEPTION, 1, str(Unit.COUNT.value))
-                        metrics.put_metric(MetricLabels.FEATURE_ERROR, 1, str(Unit.COUNT.value))
-                    logger.error("There was a problem adding features: {}".format(err))
-                    raise AddFeaturesException("Failed to add features for tile!") from err
+                # Write batch to the table and check that it succeeded for all items
+                self.batch_write_items(items)
+
+            except Exception as err:
+                if isinstance(metrics, MetricsLogger):
+                    metrics.put_metric(MetricLabels.FEATURE_UPDATE_EXCEPTION, 1, str(Unit.COUNT.value))
+                    metrics.put_metric(MetricLabels.FEATURE_ERROR, 1, str(Unit.COUNT.value))
+                raise AddFeaturesException("Failed to add features for tile!") from err
 
     @metric_scope
     def get_features(self, image_id: str, metrics: MetricsLogger = None) -> List[Feature]:
         """
-        Query the database for all items with given image_id, the convert them into feature items, then
-        go through all of the items and group the features per tile
+        Parallelized version to query the database for all items with a given image_id,
+        then convert them into feature items, and group the features per tile.
 
-        :param image_id: str = unique image_id for the job
-        :param dedupe: Optional[bool] = remove any duplicate items
+        :param image_id: The image_id to aggregate features from DDB for.
         :param metrics: MetricsLogger = the metrics logger to use to report metrics.
-
-        :return: List[Feature] = the list of features
+        :return: List of features aggregates from the DDB table.
         """
+
+        def process_query(index: int):
+            items: List[FeatureItem] = []
+            rows = self.query_items(FeatureItem(image_id + "-" + str(index)))
+            for row in rows:
+                items.append(from_dict(FeatureItem, row))
+            return items
+
         if isinstance(metrics, MetricsLogger):
             metrics.set_dimensions()
+
+        features: List[Feature] = []
+
         with Timer(
             task_str="Aggregate image features",
             metric_name=MetricLabels.FEATURE_AGG_LATENCY,
             logger=logger,
             metrics_logger=metrics,
         ):
-            # Query the database for all items with the given image_id (hash_key)
-            rows = self.query_items(FeatureItem(image_id))
+            with ThreadPoolExecutor(max_workers=5) as executor:
+                # Create a range of tasks to query the database for the salted image hash
+                futures = [executor.submit(process_query, i) for i in range(1, self.hash_salt + 1)]
+
+                # For each of the salted index processes the returned items
+                for future in as_completed(futures):
+                    feature_items = future.result()
+                    grouped_items = self.group_items_by_tile_id(feature_items)
+                    for group in grouped_items:
+                        batch_features = []
+                        for item in grouped_items[group]:
+                            if item.features:
+                                for feature in item.features:
+                                    batch_features.append(geojson.loads(feature))
+                            else:
+                                logger.warning(f"Found FeatureTable item: {item.range_key} with no features!")
+                        features.extend(batch_features)
 
-            # Convert them into feature items
-            feature_items: List[FeatureItem] = []
-            for row in rows:
-                feature_items.append(from_dict(FeatureItem, row))
-
-            # Combine all feature batches from the same tile together
-            grouped_items = self.group_items_by_tile_id(feature_items)
-
-            # Go through our items and group our features per tile
-            features: List[Feature] = []
-            for group in grouped_items:
-                batch_features = []
-                for item in grouped_items[group]:
-                    if item.features:
-                        for feature in item.features:
-                            batch_features.append(geojson.loads(feature))
-                    else:
-                        logger.warning(f"Found FeatureTable item: {item.range_key} with no features!")
-                features.extend(batch_features)
         return features
 
     def group_features_by_key(self, features: List[Feature]) -> Dict[str, List[Feature]]:
         """
         Group all the feature items by key
 
-        :param features: List[Feature] = the list of features
+        :param features: The list of features
 
-        :return: Dict[str, List[Feature]] = dict which contains a unique key and within
-                                    key contains a list of features
+        :return: Map of keys containing a list of features
         """
         result: Dict[str, List[Feature]] = {}
         for feature in features:
@@ -199,10 +220,8 @@ def group_items_by_tile_id(items: List[FeatureItem]) -> Dict[str, List[FeatureIt
         """
         Group all the feature items by tile id
 
-        :param items: List[FeatureItem] = the list of feature items
-
-        :return: Dict[str, List[FeatureItem]] = dict which contains a unique tile id and within
-                                    tile id contains a list of features
+        :param items: The list of feature items
+        :return: A unique tile id and within tile id contains a list of features
         """
         grouped_items: Dict[str, List[FeatureItem]] = {}
         for item in items:
@@ -216,13 +235,11 @@ def generate_tile_key(self, feature: Feature) -> str:
         """
         Generate the tile key based on the given feature.
 
-        :param feature: Feature = properties of a feature
-
-        :return: str = tile key
+        :param feature: Properties of a feature
+        :return: The tile key associated with this list of features.
         """
         bbox = feature["properties"][GeojsonDetectionField.BOUNDS]
 
-        # TODO: Check tile size to see if it is w,h or row/col
         # This is the size of the unique pixels in each tile
         stride_x = self.tile_size[0] - self.overlap[0]
         stride_y = self.tile_size[1] - self.overlap[1]
diff --git a/src/aws/osml/model_runner/database/job_table.py b/src/aws/osml/model_runner/database/job_table.py
index 15a039bc..0e1e2921 100755
--- a/src/aws/osml/model_runner/database/job_table.py
+++ b/src/aws/osml/model_runner/database/job_table.py
@@ -32,7 +32,7 @@ class JobItem(DDBItem):
     region_count: Optional[Decimal] = total count of regions expected for this image
     width: Optional[Decimal] = width of the image
     height: Optional[Decimal] = height of the image
-    feature_selection_options: Optional[str] = the options used in selecting features (NMS/SOFT_NMS, thresholds, etc.)
+    feature_distillation_options: Optional[str] = the options used in selecting features (NMS/SOFT_NMS, thresholds, etc.)
     """
 
     image_id: str
@@ -57,7 +57,7 @@ class JobItem(DDBItem):
     processing_time: Optional[Decimal] = None
     feature_properties: Optional[str] = None
     image_security_classification: Optional[str] = None
-    feature_selection_options: Optional[str] = None
+    feature_distillation_option: Optional[str] = None
 
     def __post_init__(self):
         self.ddb_key = DDBKey(hash_key="image_id", hash_value=self.image_id)
@@ -109,7 +109,7 @@ def start_image_request(self, image_request_item: JobItem) -> JobItem:
         except Exception as err:
             raise StartImageException("Failed to start image processing!") from err
 
-    def complete_region_request(self, image_id: str, error: bool = False) -> JobItem:
+    def complete_region_request(self, image_id: str, error: bool) -> JobItem:
         """
         Update the image job to reflect that a region has succeeded or failed.
 
diff --git a/src/aws/osml/model_runner/database/region_request_table.py b/src/aws/osml/model_runner/database/region_request_table.py
index 2fc72cbb..3491a176 100755
--- a/src/aws/osml/model_runner/database/region_request_table.py
+++ b/src/aws/osml/model_runner/database/region_request_table.py
@@ -4,12 +4,11 @@
 import time
 from dataclasses import dataclass
 from decimal import Decimal
-from enum import auto
 from typing import Optional
 
 from dacite import from_dict
 
-from aws.osml.model_runner.common import AutoStringEnum
+from aws.osml.model_runner.common import RegionRequestStatus
 
 from .ddb_helper import DDBHelper, DDBItem, DDBKey
 from .exceptions import CompleteRegionException, GetRegionRequestItemException, StartRegionException, UpdateRegionException
@@ -17,18 +16,6 @@
 logger = logging.getLogger(__name__)
 
 
-class RegionRequestStatus(str, AutoStringEnum):
-    """
-    Enumeration defining status for region
-    """
-
-    STARTING = auto()
-    PARTIAL = auto()
-    IN_PROGRESS = auto()
-    SUCCESS = auto()
-    FAILED = auto()
-
-
 @dataclass
 class RegionRequestItem(DDBItem):
     """
@@ -106,23 +93,18 @@ def start_region_request(self, region_request_item: RegionRequestItem) -> Region
         except Exception as err:
             raise StartRegionException("Failed to add region request to the table!") from err
 
-    def complete_region_request(self, region_request_item: RegionRequestItem, error=False):
+    def complete_region_request(self, region_request_item: RegionRequestItem, region_status: RegionRequestStatus):
         """
         Update the region job to reflect that a region has succeeded or failed.
 
         :param region_request_item: RegionRequestItem = the unique identifier for the region we want to update
-        :param error: bool = if there was an error processing the region, is true else false
+        :param region_status: RegionRequestStatus = Status of region at completion (FAILURE, PARTIAL, SUCCESS, etc.)
 
         :return: RegionRequestItem = Updated region request item
         """
         try:
             region_request_item.last_updated_time = Decimal(time.time() * 1000)
-
-            if error:
-                region_request_item.region_status = RegionRequestStatus.FAILED
-            else:
-                region_request_item.region_status = RegionRequestStatus.SUCCESS
-
+            region_request_item.region_status = region_status
             region_request_item.end_time = Decimal(time.time() * 1000)
 
             return from_dict(
diff --git a/src/aws/osml/model_runner/inference/exceptions.py b/src/aws/osml/model_runner/inference/exceptions.py
index cc9c926b..e561b694 100755
--- a/src/aws/osml/model_runner/inference/exceptions.py
+++ b/src/aws/osml/model_runner/inference/exceptions.py
@@ -5,5 +5,5 @@
 # flake8: noqa
 
 
-class FeatureSelectionException(Exception):
+class FeatureDistillationException(Exception):
     pass
diff --git a/src/aws/osml/model_runner/inference/feature_selection.py b/src/aws/osml/model_runner/inference/feature_selection.py
index a3df756f..e6aaa958 100755
--- a/src/aws/osml/model_runner/inference/feature_selection.py
+++ b/src/aws/osml/model_runner/inference/feature_selection.py
@@ -6,8 +6,12 @@
 from ensemble_boxes import nms, soft_nms
 from geojson import Feature
 
-from aws.osml.model_runner.common import FeatureSelectionAlgorithm, FeatureSelectionOptions, GeojsonDetectionField
-from aws.osml.model_runner.inference.exceptions import FeatureSelectionException
+from aws.osml.model_runner.common import (
+    FeatureDistillationAlgorithm,
+    FeatureDistillationAlgorithmType,
+    GeojsonDetectionField,
+)
+from aws.osml.model_runner.inference.exceptions import FeatureDistillationException
 
 
 class FeatureSelector:
@@ -18,12 +22,11 @@ class FeatureSelector:
     is instantiated.
     """
 
-    def __init__(self, options: FeatureSelectionOptions = None) -> None:
+    def __init__(self, options: FeatureDistillationAlgorithm = None) -> None:
         """
-        :param options: FeatureSelectionOptions = options to use to set the algorithm, threshods, and parameters.
-                                                    if not provided the default FeatureSelectionOptions is used
+        :param options: FeatureSelectionOptions = options to use to set the algorithm, thresholds, and parameters.
         """
-        self.options = options if options else FeatureSelectionOptions()
+        self.options = options
 
     def select_features(self, feature_list: List[Feature]) -> List[Feature]:
         """
@@ -32,8 +35,10 @@ def select_features(self, feature_list: List[Feature]) -> List[Feature]:
         """
         if feature_list is None or not feature_list:
             return []
+        if not self.options:
+            return feature_list
         boxes_list, scores_list, labels_list = self._get_lists_from_features(feature_list)
-        if self.options.algorithm == FeatureSelectionAlgorithm.SOFT_NMS:
+        if self.options.algorithm_type == FeatureDistillationAlgorithmType.SOFT_NMS:
             boxes, scores, labels = soft_nms(
                 [boxes_list],
                 [scores_list],
@@ -43,14 +48,12 @@ def select_features(self, feature_list: List[Feature]) -> List[Feature]:
                 sigma=self.options.sigma,
                 thresh=self.options.skip_box_threshold,
             )
-        elif self.options.algorithm == FeatureSelectionAlgorithm.NONE:
-            return feature_list
-        elif self.options.algorithm == FeatureSelectionAlgorithm.NMS:
+        elif self.options.algorithm_type == FeatureDistillationAlgorithmType.NMS:
             boxes, scores, labels = nms(
                 [boxes_list], [scores_list], [labels_list], weights=None, iou_thr=self.options.iou_threshold
             )
         else:
-            raise FeatureSelectionException("Invalid selection algorithm")
+            raise FeatureDistillationException(f"Invalid feature distillation algorithm: {self.options.algorithm_type}")
         return self._get_features_from_lists(boxes, scores, labels)
 
     def _get_lists_from_features(self, feature_list: List[Feature]) -> Tuple[List, List, List]:
@@ -129,7 +132,7 @@ def _get_features_from_lists(self, boxes: List, scores: List, labels: List) -> L
             feature_hash_id = hash(str(box) + category)
             feature = self.feature_id_map.get(feature_hash_id)
             if feature:
-                if self.options.algorithm == FeatureSelectionAlgorithm.SOFT_NMS and score != feature.get(
+                if self.options.algorithm_type == FeatureDistillationAlgorithmType.SOFT_NMS and score != feature.get(
                     "properties", {}
                 ).get("detection_score"):
                     feature["properties"]["adjusted_feature_types"] = {category: score}
diff --git a/src/aws/osml/model_runner/inference/feature_utils.py b/src/aws/osml/model_runner/inference/feature_utils.py
index 34e44b66..a58ea4cc 100755
--- a/src/aws/osml/model_runner/inference/feature_utils.py
+++ b/src/aws/osml/model_runner/inference/feature_utils.py
@@ -220,8 +220,8 @@ def get_source_property(image_extension: str, dataset: gdal.Dataset) -> Optional
     """
     # Currently we only support deriving source metadata from NITF images
     if image_extension == "NITF":
-        metadata = dataset.GetMetadata()
         try:
+            metadata = dataset.GetMetadata()
             # Extract metadata headers from NITF
             data_type = metadata.get("NITF_ICAT", None)
             source_id = metadata.get("NITF_FTITLE", None)
@@ -256,8 +256,8 @@ def get_source_property(image_extension: str, dataset: gdal.Dataset) -> Optional
 
             return source_property
         except Exception as err:
-            logging.warning(f"Source metadata not available for {image_extension} image extension! {err}")
+            logging.warning("Source metadata not available for {} image extension! {}".format(image_extension, err))
             return None
     else:
-        logging.warning(f"Source metadata not available for {image_extension} image extension!")
+        logging.warning("Source metadata not available for {} image extension!".format(image_extension))
         return None
diff --git a/src/aws/osml/model_runner/inference/http_detector.py b/src/aws/osml/model_runner/inference/http_detector.py
index 1b3daf61..2d9b91bf 100644
--- a/src/aws/osml/model_runner/inference/http_detector.py
+++ b/src/aws/osml/model_runner/inference/http_detector.py
@@ -125,7 +125,9 @@ def find_features(self, payload: BufferedReader, metrics: MetricsLogger) -> Feat
                 metrics_logger=metrics,
             ):
                 # If we are not running against a real model
-                if self.endpoint == ServiceConfig.noop_model_name:
+                if self.endpoint == ServiceConfig.noop_geom_model_name:
+                    return create_mock_feature_collection(payload, geom=True)
+                elif self.endpoint == ServiceConfig.noop_bounds_model_name:
                     return create_mock_feature_collection(payload)
                 else:
                     response = self.http_pool.request(
diff --git a/src/aws/osml/model_runner/tile_worker/tile_worker.py b/src/aws/osml/model_runner/tile_worker/tile_worker.py
index de972d80..76b2d4d0 100755
--- a/src/aws/osml/model_runner/tile_worker/tile_worker.py
+++ b/src/aws/osml/model_runner/tile_worker/tile_worker.py
@@ -90,12 +90,13 @@ def run(self) -> None:
                         # Create a geometry for each feature in the result. The geographic coordinates of these
                         # features are computed using the sensor model provided in the image metadata
                         self.feature_refinery.refine_features_for_tile(features)
-                        logging.info(f"Created Geographic Coordinates for {len(features)} features")
+                        logging.info("Created Geographic Coordinates for {} features".format(len(features)))
 
                     self.feature_table.add_features(features)
             except Exception as e:
                 logging.error("Failed to process region tile!")
                 logging.exception(e)
+                self.feature_detector.error_count += 1  # borrow the feature detector error count to tally other errors
                 if self.metrics:
                     self.metrics.put_metric(MetricLabels.TILE_PROCESSING_ERROR, 1, str(Unit.COUNT.value))
                     self.metrics.put_metric(MetricLabels.TILE_CREATION_FAILURE, 1, str(Unit.COUNT.value))
diff --git a/src/aws/osml/model_runner/tile_worker/tile_worker_utils.py b/src/aws/osml/model_runner/tile_worker/tile_worker_utils.py
index af02b5d6..82a4426a 100755
--- a/src/aws/osml/model_runner/tile_worker/tile_worker_utils.py
+++ b/src/aws/osml/model_runner/tile_worker/tile_worker_utils.py
@@ -79,7 +79,7 @@ def setup_tile_workers(
 
         return tile_queue, tile_workers
     except Exception as err:
-        logger.exception("Failed to setup tile workers!: {}", err)
+        logger.exception("Failed to setup tile workers!: {}".format(err))
         raise SetupTileWorkersException("Failed to setup tile workers!") from err
 
 
@@ -90,7 +90,7 @@ def process_tiles(
     raster_dataset: gdal.Dataset,
     metrics: MetricsLogger = None,
     sensor_model: Optional[SensorModel] = None,
-) -> int:
+) -> Tuple[int, int]:
     """
     Loads a GDAL dataset into memory and processes it with a pool of tile workers.
 
@@ -101,7 +101,7 @@ def process_tiles(
     :param metrics: MetricsLogger = the metrics logger to use to report metrics.
     :param sensor_model: Optional[SensorModel] = the sensor model for this raster dataset
 
-    :return: int = number of tiles processed
+    :return: Tuple[int, int] = number of tiles processed, number of tiles with an error
     """
     try:
         # This will update the GDAL configuration options to use the security credentials for
@@ -135,7 +135,7 @@ def process_tiles(
                     region_request.tile_size,
                     region_request.tile_overlap,
                 ):
-                    # Create a temp file name for the NITF encoded region
+                    # Create a temp file name for the encoded region
                     region_image_filename = "{}-region-{}-{}-{}-{}.{}".format(
                         token_hex(16),
                         tile_bounds[0][0],
@@ -202,19 +202,21 @@ def process_tiles(
                 # the temp directory. If the context is exited before all workers return then
                 # the directory will be deleted, and we will potentially lose tiles.
                 # Wait for all the workers to finish gracefully before we clean up the temp directory
+                tile_error_count = 0
                 for worker in tile_workers:
                     worker.join()
+                    tile_error_count += worker.feature_detector.error_count
 
         logger.info(
-            "Model Runner Stats Processed {} image tiles for region {}.".format(
-                total_tile_count, region_request.region_bounds
+            "Model Runner Stats Processed {} image tiles for region {}. {} tile errors.".format(
+                total_tile_count, region_request.region_bounds, tile_error_count
             )
         )
     except Exception as err:
         logger.exception("File processing tiles: {}", err)
         raise ProcessTilesException("Failed to process tiles!") from err
 
-    return total_tile_count
+    return total_tile_count, tile_error_count
 
 
 def sizeof_fmt(num: float, suffix: str = "B") -> str:


Test patch:
diff --git a/test/aws/osml/model_runner/common/test_feature_selection_options.py b/test/aws/osml/model_runner/common/test_feature_selection_options.py
deleted file mode 100755
index 8dfb9f84..00000000
--- a/test/aws/osml/model_runner/common/test_feature_selection_options.py
+++ /dev/null
@@ -1,32 +0,0 @@
-#  Copyright 2023 Amazon.com, Inc. or its affiliates.
-
-from dataclasses import asdict
-from json import dumps, loads
-from unittest import TestCase
-
-from dacite import Config, from_dict
-
-
-class TestFeatureSelectionOptions(TestCase):
-    def test_feature_feature_selection_options_serialization(self):
-        from aws.osml.model_runner.common import (
-            FeatureSelectionAlgorithm,
-            FeatureSelectionOptions,
-            feature_selection_options_factory,
-        )
-
-        feature_selection_options = FeatureSelectionOptions()
-
-        feature_selection_options_json = dumps(
-            asdict(feature_selection_options, dict_factory=feature_selection_options_factory)
-        )
-        expected_json = '{"algorithm": "NMS", "iou_threshold": 0.75, "skip_box_threshold": 0.0001, "sigma": 0.1}'
-        assert feature_selection_options_json == expected_json
-
-        new_feature_selection_options = from_dict(
-            data_class=FeatureSelectionOptions,
-            data=loads(feature_selection_options_json),
-            config=Config(cast=[FeatureSelectionAlgorithm]),
-        )
-        assert isinstance(new_feature_selection_options, FeatureSelectionOptions)
-        assert isinstance(new_feature_selection_options.algorithm, FeatureSelectionAlgorithm)
diff --git a/test/aws/osml/model_runner/common/test_mr_post_processing.py b/test/aws/osml/model_runner/common/test_mr_post_processing.py
new file mode 100755
index 00000000..a678c72b
--- /dev/null
+++ b/test/aws/osml/model_runner/common/test_mr_post_processing.py
@@ -0,0 +1,118 @@
+#  Copyright 2023 Amazon.com, Inc. or its affiliates.
+
+from dataclasses import asdict
+from json import dumps, loads
+from unittest import TestCase
+
+
+class TestMRPostProcessing(TestCase):
+    def test_feature_distillation_deserializer_nms(self):
+        from aws.osml.model_runner.common import (
+            FeatureDistillationAlgorithmType,
+            FeatureDistillationDeserializer,
+            FeatureDistillationNMS,
+            mr_post_processing_options_factory,
+        )
+
+        feature_selection_option = FeatureDistillationNMS()
+
+        feature_selection_options_json = dumps(
+            asdict(feature_selection_option, dict_factory=mr_post_processing_options_factory)
+        )
+        expected_json = '{"algorithm_type": "NMS", "iou_threshold": 0.75}'
+        assert feature_selection_options_json == expected_json
+
+        deserializer = FeatureDistillationDeserializer()
+        new_feature_selection_option = deserializer.deserialize(loads(feature_selection_options_json))
+
+        assert isinstance(new_feature_selection_option, FeatureDistillationNMS)
+        assert isinstance(new_feature_selection_option.algorithm_type, FeatureDistillationAlgorithmType)
+
+    def test_feature_distillation_deserializer_soft_nms(self):
+        from aws.osml.model_runner.common import (
+            FeatureDistillationAlgorithmType,
+            FeatureDistillationDeserializer,
+            FeatureDistillationSoftNMS,
+            mr_post_processing_options_factory,
+        )
+
+        feature_selection_option = FeatureDistillationSoftNMS()
+
+        feature_selection_options_json = dumps(
+            asdict(feature_selection_option, dict_factory=mr_post_processing_options_factory)
+        )
+        expected_json = '{"algorithm_type": "SOFT_NMS", "iou_threshold": 0.75, "skip_box_threshold": 0.0001, "sigma": 0.1}'
+        assert feature_selection_options_json == expected_json
+
+        deserializer = FeatureDistillationDeserializer()
+        new_feature_selection_option = deserializer.deserialize(loads(feature_selection_options_json))
+
+        assert isinstance(new_feature_selection_option, FeatureDistillationSoftNMS)
+        assert isinstance(new_feature_selection_option.algorithm_type, FeatureDistillationAlgorithmType)
+
+    def test_feature_distillation_deserializer_missing_algorithm_type(self):
+        from aws.osml.model_runner.common import FeatureDistillationDeserializer
+
+        invalid_json = '{"iou_threshold": 0.75, "skip_box_threshold": 0.0001}'
+
+        deserializer = FeatureDistillationDeserializer()
+        with self.assertRaises(ValueError):
+            deserializer.deserialize(loads(invalid_json))
+
+    def test_feature_distillation_deserializer_invalid_algorithm_type(self):
+        from aws.osml.model_runner.common import FeatureDistillationDeserializer
+
+        invalid_json = '{"algorithm_type": "MAGIC", "iou_threshold": 0.75, "skip_box_threshold": 0.0001}'
+
+        deserializer = FeatureDistillationDeserializer()
+        with self.assertRaises(ValueError):
+            deserializer.deserialize(loads(invalid_json))
+
+    def test_mr_post_processing_deserializer_feature_distillation_step(self):
+        from aws.osml.model_runner.common import (
+            FeatureDistillationDeserializer,
+            MRPostProcessingDeserializer,
+            MRPostprocessingStep,
+        )
+
+        step = MRPostprocessingStep.FEATURE_DISTILLATION
+
+        deserializer = MRPostProcessingDeserializer.get_deserializer(step)
+        assert isinstance(deserializer, FeatureDistillationDeserializer)
+
+    def test_mr_post_processing_deserializer_invalid_step(self):
+        from enum import auto
+
+        from aws.osml.model_runner.common import MRPostProcessingDeserializer
+
+        class MRPostprocessingStep:
+            MAGIC = auto()
+
+        step = MRPostprocessingStep.MAGIC
+        with self.assertRaises(ValueError):
+            MRPostProcessingDeserializer.get_deserializer(step)
+
+    def test_deserialize_post_processing_list(self):
+        from aws.osml.model_runner.common import MRPostProcessing, deserialize_post_processing_list
+
+        test_list = [
+            {"step": "FEATURE_DISTILLATION", "algorithm": {"algorithm_type": "NMS", "iou_threshold": 0.75}},
+            {
+                "step": "FEATURE_DISTILLATION",
+                "algorithm": {"algorithm_type": "SOFT_NMS", "iou_threshold": 0.75, "skip_box_threshold": 0.0001},
+            },
+        ]
+        deserialized_list = deserialize_post_processing_list(test_list)
+
+        assert isinstance(deserialized_list[0], MRPostProcessing)
+        assert isinstance(deserialized_list[1], MRPostProcessing)
+
+    def test_deserialize_post_processing_list_exception(self):
+        from aws.osml.model_runner.common import deserialize_post_processing_list
+
+        bad_test_list = [
+            {"step": "FEATURE_DISTILLATION", "algorithm": {"algorithm_type": "NMS", "iou_threshold": 0.75}},
+            {"step": "FIND_ALL_THE_OBJECTS", "algorithm": {"algorithm_type": "MAGIC"}},
+        ]
+        with self.assertRaises(Exception):
+            deserialize_post_processing_list(bad_test_list)
diff --git a/test/aws/osml/model_runner/database/test_feature_table.py b/test/aws/osml/model_runner/database/test_feature_table.py
index 883f901e..20d986a7 100755
--- a/test/aws/osml/model_runner/database/test_feature_table.py
+++ b/test/aws/osml/model_runner/database/test_feature_table.py
@@ -57,7 +57,9 @@
     },
 }
 
-TEST_MOCK_PUT_EXCEPTION = Mock(side_effect=ClientError({"Error": {"Code": 500, "Message": "ClientError"}}, "put_item"))
+TEST_MOCK_BATCH_WRITE_EXCEPTION = Mock(
+    side_effect=ClientError({"Error": {"Code": 500, "Message": "ClientError"}}, "batch_write_item")
+)
 
 
 @mock_dynamodb
@@ -79,6 +81,7 @@ def setUp(self):
             BillingMode="PAY_PER_REQUEST",
         )
         self.feature_table = FeatureTable(os.environ["FEATURE_TABLE"], (2048, 2048), (50, 50))
+        self.feature_table.hash_salt = 1
 
     def tearDown(self):
         """
@@ -140,7 +143,7 @@ def test_add_features_throw_exceptions(self):
         from aws.osml.model_runner.database.exceptions import AddFeaturesException
 
         features = self.get_feature_list()
-        self.feature_table.table.put_item = TEST_MOCK_PUT_EXCEPTION
+        self.feature_table.batch_write_items = TEST_MOCK_BATCH_WRITE_EXCEPTION
         with self.assertRaises(AddFeaturesException):
             self.feature_table.add_features(features)
 
diff --git a/test/aws/osml/model_runner/database/test_job_table.py b/test/aws/osml/model_runner/database/test_job_table.py
index 0f8a5b45..5fd16fb7 100755
--- a/test/aws/osml/model_runner/database/test_job_table.py
+++ b/test/aws/osml/model_runner/database/test_job_table.py
@@ -61,7 +61,7 @@ def test_region_complete_success_count(self):
         Validate that when we complete a region successfully it updates the ddb item
         """
         self.job_table.start_image_request(self.job_item)
-        self.job_table.complete_region_request(TEST_IMAGE_ID)
+        self.job_table.complete_region_request(TEST_IMAGE_ID, False)
         resulting_job_item = self.job_table.get_image_request(TEST_IMAGE_ID)
         assert resulting_job_item.region_success == Decimal(1)
         assert resulting_job_item.region_error == Decimal(0)
@@ -118,7 +118,7 @@ def test_complete_region_failure(self):
         self.job_table.table.update_item = TEST_MOCK_UPDATE_EXCEPTION
         self.job_table.start_image_request(self.job_item)
         with self.assertRaises(CompleteRegionException):
-            self.job_table.complete_region_request(TEST_IMAGE_ID)
+            self.job_table.complete_region_request(TEST_IMAGE_ID, False)
 
     def test_end_image_failure(self):
         from aws.osml.model_runner.database.exceptions import EndImageException
diff --git a/test/aws/osml/model_runner/database/test_region_request_table.py b/test/aws/osml/model_runner/database/test_region_request_table.py
index b4b8f264..1e7e8e43 100755
--- a/test/aws/osml/model_runner/database/test_region_request_table.py
+++ b/test/aws/osml/model_runner/database/test_region_request_table.py
@@ -57,7 +57,7 @@ def test_region_started_success(self):
         """
         Validate we can start a region, and it gets created in the table
         """
-        from aws.osml.model_runner.database.region_request_table import RegionRequestStatus
+        from aws.osml.model_runner.common import RegionRequestStatus
 
         self.region_request_table.start_region_request(self.region_request_item)
         resulting_region_request_item = self.region_request_table.get_region_request(TEST_REGION_ID, TEST_IMAGE_ID)
@@ -70,10 +70,10 @@ def test_region_complete_success(self):
         """
         Validate that when we complete a region successfully it updates the ddb item
         """
-        from aws.osml.model_runner.database.region_request_table import RegionRequestStatus
+        from aws.osml.model_runner.common import RegionRequestStatus
 
         self.region_request_table.start_region_request(self.region_request_item)
-        self.region_request_table.complete_region_request(self.region_request_item)
+        self.region_request_table.complete_region_request(self.region_request_item, RegionRequestStatus.SUCCESS)
         resulting_region_request_item = self.region_request_table.get_region_request(TEST_REGION_ID, TEST_IMAGE_ID)
 
         assert resulting_region_request_item.region_status == RegionRequestStatus.SUCCESS
@@ -96,10 +96,10 @@ def test_region_complete_failed(self):
         """
         Validate that when we complete a region successfully it updates the ddb item
         """
-        from aws.osml.model_runner.database.region_request_table import RegionRequestStatus
+        from aws.osml.model_runner.common import RegionRequestStatus
 
         self.region_request_table.start_region_request(self.region_request_item)
-        self.region_request_table.complete_region_request(self.region_request_item, True)
+        self.region_request_table.complete_region_request(self.region_request_item, RegionRequestStatus.FAILED)
         resulting_region_request_item = self.region_request_table.get_region_request(TEST_REGION_ID, TEST_IMAGE_ID)
 
         assert resulting_region_request_item.region_status == RegionRequestStatus.FAILED
diff --git a/test/aws/osml/model_runner/inference/test_feature_selection.py b/test/aws/osml/model_runner/inference/test_feature_selection.py
index d0b41545..a18a84f2 100755
--- a/test/aws/osml/model_runner/inference/test_feature_selection.py
+++ b/test/aws/osml/model_runner/inference/test_feature_selection.py
@@ -8,34 +8,31 @@
 
 class TestFeatureSelection(TestCase):
     def test_feature_selection_empty_list(self):
-        from aws.osml.model_runner.common import FeatureSelectionAlgorithm, FeatureSelectionOptions
+        from aws.osml.model_runner.common import FeatureDistillationNMS
         from aws.osml.model_runner.inference import FeatureSelector
 
-        feature_selection_options = FeatureSelectionOptions()
-        feature_selection_options.algorithm = FeatureSelectionAlgorithm.NMS
-        feature_selector = FeatureSelector(options=feature_selection_options)
+        feature_selection_option = FeatureDistillationNMS()
+        feature_selector = FeatureSelector(options=feature_selection_option)
 
         features = feature_selector.select_features([])
         assert len(features) == 0
 
     def test_feature_selection_none_list(self):
-        from aws.osml.model_runner.common import FeatureSelectionAlgorithm, FeatureSelectionOptions
+        from aws.osml.model_runner.common import FeatureDistillationNMS
         from aws.osml.model_runner.inference import FeatureSelector
 
-        feature_selection_options = FeatureSelectionOptions()
-        feature_selection_options.algorithm = FeatureSelectionAlgorithm.NMS
-        feature_selector = FeatureSelector(options=feature_selection_options)
+        feature_selection_option = FeatureDistillationNMS()
+        feature_selector = FeatureSelector(options=feature_selection_option)
 
         features = feature_selector.select_features(None)
         assert len(features) == 0
 
     def test_feature_selection_nms_no_overlap(self):
-        from aws.osml.model_runner.common import FeatureSelectionAlgorithm, FeatureSelectionOptions
+        from aws.osml.model_runner.common import FeatureDistillationNMS
         from aws.osml.model_runner.inference import FeatureSelector
 
-        feature_selection_options = FeatureSelectionOptions()
-        feature_selection_options.algorithm = FeatureSelectionAlgorithm.NMS
-        feature_selector = FeatureSelector(options=feature_selection_options)
+        feature_selection_option = FeatureDistillationNMS()
+        feature_selector = FeatureSelector(options=feature_selection_option)
 
         with open("./test/data/detections.geojson", "r") as geojson_file:
             sample_features = geojson.load(geojson_file)["features"]
@@ -45,12 +42,11 @@ def test_feature_selection_nms_no_overlap(self):
         assert len(sample_features) == len(processed_features)
 
     def test_feature_selection_nms_overlaps(self):
-        from aws.osml.model_runner.common import FeatureSelectionAlgorithm, FeatureSelectionOptions
+        from aws.osml.model_runner.common import FeatureDistillationNMS
         from aws.osml.model_runner.inference import FeatureSelector
 
-        feature_selection_options = FeatureSelectionOptions()
-        feature_selection_options.algorithm = FeatureSelectionAlgorithm.NMS
-        feature_selector = FeatureSelector(options=feature_selection_options)
+        feature_selection_option = FeatureDistillationNMS()
+        feature_selector = FeatureSelector(options=feature_selection_option)
 
         # The actual geojson isn't used by nms. The only thing we care about for determining
         # duplicates is the bounds_imcoords property so these shapes are all just simple
@@ -77,12 +73,9 @@ def test_feature_selection_nms_overlaps(self):
         assert len(processed_features) == 2
 
     def test_feature_selection_no_selection(self):
-        from aws.osml.model_runner.common import FeatureSelectionAlgorithm, FeatureSelectionOptions
         from aws.osml.model_runner.inference import FeatureSelector
 
-        feature_selection_options = FeatureSelectionOptions()
-        feature_selection_options.algorithm = FeatureSelectionAlgorithm.NONE
-        feature_selector = FeatureSelector(options=feature_selection_options)
+        feature_selector = FeatureSelector()
 
         # feature_a and feature_b have IoU ~0.80
         original_features = [
@@ -106,18 +99,15 @@ def test_feature_selection_no_selection(self):
         assert processed_features == original_features
 
     def test_feature_selection_unknown_algorithm(self):
-        from enum import Enum
-
-        from aws.osml.model_runner.common import FeatureSelectionOptions
+        from aws.osml.model_runner.common import FeatureDistillationNMS, MRPostProcessingAlgorithmType
         from aws.osml.model_runner.inference import FeatureSelector
-        from aws.osml.model_runner.inference.exceptions import FeatureSelectionException
+        from aws.osml.model_runner.inference.exceptions import FeatureDistillationException
 
-        class FakeFeatureSelectionAlgorithm(str, Enum):
+        class FakeFeatureSelectionAlgorithm(MRPostProcessingAlgorithmType):
             MAGIC = "MAGIC"
 
-        feature_selection_options = FeatureSelectionOptions()
-        feature_selection_options.algorithm = FakeFeatureSelectionAlgorithm.MAGIC
-        feature_selector = FeatureSelector(options=feature_selection_options)
+        feature_selection_option = FeatureDistillationNMS(algorithm_type=FakeFeatureSelectionAlgorithm.MAGIC)
+        feature_selector = FeatureSelector(options=feature_selection_option)
 
         # feature_a and feature_b have IoU ~0.80
         original_features = [
@@ -137,21 +127,17 @@ class FakeFeatureSelectionAlgorithm(str, Enum):
                 properties={"bounds_imcoords": [250, 250, 300, 275]},
             ),
         ]
-        with self.assertRaises(FeatureSelectionException):
+        with self.assertRaises(FeatureDistillationException):
             feature_selector.select_features(original_features)
 
     def test_feature_selection_nms_overlaps_custom_threshold(self):
-        from aws.osml.model_runner.common import FeatureSelectionAlgorithm, FeatureSelectionOptions
+        from aws.osml.model_runner.common import FeatureDistillationNMS
         from aws.osml.model_runner.inference import FeatureSelector
 
-        feature_selection_options_1 = FeatureSelectionOptions()
-        feature_selection_options_1.algorithm = FeatureSelectionAlgorithm.NMS
-        feature_selection_options_1.iou_threshold = 0.4
-        feature_selector_1 = FeatureSelector(options=feature_selection_options_1)
-        feature_selection_options_2 = FeatureSelectionOptions()
-        feature_selection_options_2.algorithm = FeatureSelectionAlgorithm.NMS
-        feature_selection_options_2.iou_threshold = 0.7
-        feature_selector_2 = FeatureSelector(options=feature_selection_options_2)
+        feature_selection_option_1 = FeatureDistillationNMS(iou_threshold=0.4)
+        feature_selector_1 = FeatureSelector(options=feature_selection_option_1)
+        feature_selection_option_2 = FeatureDistillationNMS(iou_threshold=0.7)
+        feature_selector_2 = FeatureSelector(options=feature_selection_option_2)
 
         # These features have an IoU of ~0.47
         original_features = [
@@ -170,11 +156,10 @@ def test_feature_selection_nms_overlaps_custom_threshold(self):
         assert len(feature_selector_2.select_features(original_features)) == 2
 
     def test_feature_selection_nms_overlaps_multiple_categories(self):
-        from aws.osml.model_runner.common import FeatureSelectionAlgorithm, FeatureSelectionOptions
+        from aws.osml.model_runner.common import FeatureDistillationNMS
         from aws.osml.model_runner.inference import FeatureSelector
 
-        feature_selection_options = FeatureSelectionOptions()
-        feature_selection_options.algorithm = FeatureSelectionAlgorithm.NMS
+        feature_selection_options = FeatureDistillationNMS()
         feature_selector = FeatureSelector(options=feature_selection_options)
 
         # feature_a and feature_b have IoU ~0.80 but different labels
@@ -208,12 +193,11 @@ def test_feature_selection_nms_overlaps_multiple_categories(self):
         assert len(processed_features) == 3
 
     def test_feature_selection_soft_nms_overlaps(self):
-        from aws.osml.model_runner.common import FeatureSelectionAlgorithm, FeatureSelectionOptions
+        from aws.osml.model_runner.common import FeatureDistillationSoftNMS
         from aws.osml.model_runner.inference import FeatureSelector
 
-        feature_selection_options = FeatureSelectionOptions()
-        feature_selection_options.algorithm = FeatureSelectionAlgorithm.SOFT_NMS
-        feature_selector = FeatureSelector(options=feature_selection_options)
+        feature_selection_option = FeatureDistillationSoftNMS()
+        feature_selector = FeatureSelector(options=feature_selection_option)
 
         # feature_a and feature_b have IoU ~0.80
         original_features = [
diff --git a/test/aws/osml/model_runner/inference/test_feature_utils.py b/test/aws/osml/model_runner/inference/test_feature_utils.py
index bd9f9733..ce3ac102 100755
--- a/test/aws/osml/model_runner/inference/test_feature_utils.py
+++ b/test/aws/osml/model_runner/inference/test_feature_utils.py
@@ -92,10 +92,21 @@ def test_calculate_processing_bounds_no_roi(self):
 
     def test_calculate_processing_bounds_full_image(self):
         from aws.osml.model_runner.inference.feature_utils import calculate_processing_bounds
+        from aws.osml.photogrammetry import ImageCoordinate
 
         ds, sensor_model = self.get_dataset_and_camera()
 
-        roi = shapely.wkt.loads("POLYGON ((8 50, 10 50, 10 60, 8 60, 8 50))")
+        chip_ul = sensor_model.image_to_world(ImageCoordinate([0, 0]))
+        chip_lr = sensor_model.image_to_world(ImageCoordinate([101, 101]))
+        min_vals = np.minimum(chip_ul.coordinate, chip_lr.coordinate)
+        max_vals = np.maximum(chip_ul.coordinate, chip_lr.coordinate)
+        polygon_coords = []
+        polygon_coords.append([degrees(min_vals[0]), degrees(min_vals[1])])
+        polygon_coords.append([degrees(min_vals[0]), degrees(max_vals[1])])
+        polygon_coords.append([degrees(max_vals[0]), degrees(max_vals[1])])
+        polygon_coords.append([degrees(max_vals[0]), degrees(min_vals[1])])
+        polygon_coords.append([degrees(min_vals[0]), degrees(min_vals[1])])
+        roi = shapely.geometry.Polygon(polygon_coords)
 
         processing_bounds = calculate_processing_bounds(ds, roi, sensor_model)
 
@@ -147,6 +158,19 @@ def test_calculate_processing_bounds_chip(self):
         # Processing bounds is in ((r, c), (w, h))
         assert processing_bounds == ((15, 10), (60, 75))
 
+    def test_get_source_property_not_available(self):
+        from aws.osml.model_runner.inference.feature_utils import get_source_property
+
+        ds, sensor_model = self.get_dataset_and_camera()
+        source_property = get_source_property("UNSUPPORTED", ds)
+        assert source_property is None
+
+    def test_get_source_property_exception(self):
+        from aws.osml.model_runner.inference.feature_utils import get_source_property
+
+        source_property = get_source_property("NITF", dataset=None)
+        assert source_property is None
+
     @staticmethod
     def build_gdal_sensor_model():
         from aws.osml.photogrammetry import GDALAffineSensorModel
diff --git a/test/aws/osml/model_runner/inference/test_http_detector.py b/test/aws/osml/model_runner/inference/test_http_detector.py
new file mode 100644
index 00000000..0bdececc
--- /dev/null
+++ b/test/aws/osml/model_runner/inference/test_http_detector.py
@@ -0,0 +1,90 @@
+import json
+from unittest import TestCase
+
+import mock
+from urllib3.response import HTTPResponse
+
+MOCK_RESPONSE = HTTPResponse(
+    body=json.dumps(
+        {
+            "type": "FeatureCollection",
+            "features": [
+                {
+                    "type": "Feature",
+                    "id": "1cc5e6d6-e12f-430d-adf0-8d2276ce8c5a",
+                    "geometry": {"type": "Point", "coordinates": [0.0, 0.0]},
+                    "properties": {
+                        "bounds_imcoords": [429, 553, 440, 561],
+                        "feature_types": {"ground_motor_passenger_vehicle": 0.2961518168449402},
+                        "detection_score": 0.2961518168449402,
+                        "image_id": "test-image-id",
+                    },
+                }
+            ],
+        }
+    ).encode(),
+    status=200,
+)
+
+MOCK_BAD_JSON_RESPONSE = HTTPResponse(body="Not a json string".encode(), status=200)
+
+
+class TestSMDetector(TestCase):
+    @mock.patch("aws.osml.model_runner.inference.http_detector.urllib3.PoolManager", autospec=True)
+    def test_find_features(self, mock_pool_manager):
+        from aws.osml.model_runner.inference import HTTPDetector
+
+        mock_endpoint = "http://dummy/endpoint"
+        mock_name = "test"
+        feature_detector = HTTPDetector(endpoint=mock_endpoint, name=mock_name)
+        assert feature_detector.name == mock_name
+
+        mock_pool_manager.return_value.request.return_value = MOCK_RESPONSE
+
+        with open("./test/data/small.ntf", "rb") as image_file:
+            feature_collection = feature_detector.find_features(image_file)
+            assert feature_collection["type"] == "FeatureCollection"
+            assert len(feature_collection["features"]) == 1
+
+    @mock.patch("aws.osml.model_runner.inference.http_detector.urllib3.PoolManager", autospec=True)
+    def test_find_features_RetryError(self, mock_pool_manager):
+        from requests.exceptions import RetryError
+
+        from aws.osml.model_runner.inference import HTTPDetector
+
+        mock_endpoint = "http://dummy/endpoint"
+        feature_detector = HTTPDetector(endpoint=mock_endpoint)
+
+        mock_pool_manager.return_value.request.side_effect = RetryError("test RetryError")
+
+        with open("./test/data/small.ntf", "rb") as image_file:
+            feature_detector.find_features(image_file)
+            assert feature_detector.error_count == 1
+
+    @mock.patch("aws.osml.model_runner.inference.http_detector.urllib3.PoolManager", autospec=True)
+    def test_find_features_MaxRetryError(self, mock_pool_manager):
+        from urllib3.exceptions import MaxRetryError
+
+        from aws.osml.model_runner.inference import HTTPDetector
+
+        mock_endpoint = "http://dummy/endpoint"
+        feature_detector = HTTPDetector(endpoint=mock_endpoint)
+
+        mock_pool_manager.return_value.request.side_effect = MaxRetryError("test MaxRetryError", url=mock_endpoint)
+
+        with open("./test/data/small.ntf", "rb") as image_file:
+            feature_detector.find_features(image_file)
+            assert feature_detector.error_count == 1
+
+    @mock.patch("aws.osml.model_runner.inference.http_detector.urllib3.PoolManager", autospec=True)
+    def test_find_features_JSONDecodeError(self, mock_pool_manager):
+        from aws.osml.model_runner.inference import HTTPDetector
+
+        mock_endpoint = "http://dummy/endpoint"
+        feature_detector = HTTPDetector(endpoint=mock_endpoint)
+
+        mock_pool_manager.return_value.request.return_value = MOCK_BAD_JSON_RESPONSE
+
+        with open("./test/data/small.ntf", "rb") as image_file:
+            feature_detector.find_features(image_file)
+            assert feature_detector.error_count == 1
diff --git a/test/aws/osml/model_runner/tile_worker/test_tile_worker_utils.py b/test/aws/osml/model_runner/tile_worker/test_tile_worker_utils.py
index b19dfe26..bdaba573 100755
--- a/test/aws/osml/model_runner/tile_worker/test_tile_worker_utils.py
+++ b/test/aws/osml/model_runner/tile_worker/test_tile_worker_utils.py
@@ -1,11 +1,72 @@
 #  Copyright 2023 Amazon.com, Inc. or its affiliates.
 
-import unittest
+from unittest import TestCase, main, mock
 
 import pytest
+from mock import Mock
+
+
+class TestTileWorkerUtils(TestCase):
+    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.TileWorker", autospec=True)
+    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.ServiceConfig", autospec=True)
+    def test_setup_tile_workers(self, mock_service_config, mock_tile_worker):
+        from aws.osml.model_runner.api import RegionRequest
+        from aws.osml.model_runner.tile_worker.tile_worker_utils import setup_tile_workers
+
+        mock_tile_worker.start = Mock()
+        mock_num_tile_workers = 4
+        mock_service_config.workers = mock_num_tile_workers
+        mock_region_request = RegionRequest(
+            {
+                "tile_size": (10, 10),
+                "tile_overlap": (1, 1),
+                "tile_format": "NITF",
+                "image_id": "1",
+                "image_url": "/mock/path",
+                "region_bounds": ((0, 0), (50, 50)),
+                "model_invoke_mode": "SM_ENDPOINT",
+                "image_extension": "fake",
+            }
+        )
+        mock_sensor_model = None
+        mock_elevation_model = None
+        mock_metrics = None
+        work_queue, tile_worker_list = setup_tile_workers(
+            mock_region_request, mock_sensor_model, mock_elevation_model, mock_metrics
+        )
+        assert len(tile_worker_list) == mock_num_tile_workers
+
+    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.FeatureTable", autospec=True)
+    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.TileWorker", autospec=True)
+    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.ServiceConfig", autospec=True)
+    def test_setup_tile_workers_exception(self, mock_service_config, mock_tile_worker, mock_feature_table):
+        from aws.osml.model_runner.api import RegionRequest
+        from aws.osml.model_runner.tile_worker.exceptions import SetupTileWorkersException
+        from aws.osml.model_runner.tile_worker.tile_worker_utils import setup_tile_workers
+
+        mock_tile_worker.start = Mock()
+        mock_num_tile_workers = 4
+        mock_service_config.workers = mock_num_tile_workers
+        mock_feature_table.side_effect = Exception("Mock processing exception")
+        mock_region_request = RegionRequest(
+            {
+                "tile_size": (10, 10),
+                "tile_overlap": (1, 1),
+                "tile_format": "NITF",
+                "image_id": "1",
+                "image_url": "/mock/path",
+                "region_bounds": ((0, 0), (50, 50)),
+                "model_invoke_mode": "SM_ENDPOINT",
+                "image_extension": "fake",
+            }
+        )
+        mock_sensor_model = None
+        mock_elevation_model = None
+        mock_metrics = None
+        with self.assertRaises(SetupTileWorkersException):
+            # with self.assertRaises(ValueError):
+            setup_tile_workers(mock_region_request, mock_sensor_model, mock_elevation_model, mock_metrics)
 
-
-class TestTileWorkerUtils(unittest.TestCase):
     def test_chip_generator(self):
         from aws.osml.model_runner.tile_worker.tile_worker_utils import generate_crops
 
@@ -127,4 +188,4 @@ def next_greater_power_of_two(n: int) -> int:
 
 
 if __name__ == "__main__":
-    unittest.main()
+    main()
diff --git a/test/test_app.py b/test/test_app.py
index efa9871b..8785e702 100755
--- a/test/test_app.py
+++ b/test/test_app.py
@@ -2,6 +2,7 @@
 import os
 import unittest
 from importlib import reload
+from queue import Queue
 
 import boto3
 import geojson
@@ -296,6 +297,18 @@ def tearDown(self):
     def test_aws_osml_model_runner_importable(self):
         import aws.osml.model_runner  # noqa: F401
 
+    def test_run(self):
+        self.model_runner.monitor_work_queues = Mock()
+
+        self.model_runner.run()
+
+        self.model_runner.monitor_work_queues.assert_called_once()
+
+    def test_stop(self):
+        self.model_runner.running = True
+        self.model_runner.stop()
+        assert self.model_runner.running is False
+
     def test_process_bounds_image_request(self):
         from aws.osml.model_runner.database.region_request_table import RegionRequestTable
 
@@ -401,16 +414,12 @@ def test_process_geom_image_request(self):
     # Remember that with multiple patch decorators the order of the mocks in the parameter list is
     # reversed (i.e. the first mock parameter is the last decorator defined). Also note that the
     # pytest fixtures must come at the end.
-    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.FeatureDetectorFactory", autospec=True)
-    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.FeatureTable", autospec=True)
-    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.TileWorker", autospec=True)
-    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.Queue", autospec=True)
+    @mock.patch("aws.osml.model_runner.app.setup_tile_workers")
+    @mock.patch("aws.osml.model_runner.app.process_tiles")
     def test_process_region_request(
         self,
-        mock_queue,
-        mock_tile_worker,
-        mock_feature_table,
-        mock_feature_detector,
+        mock_process_tiles,
+        mock_setup_tile_workers,
     ):
         from aws.osml.gdal.gdal_utils import load_gdal_dataset
         from aws.osml.model_runner.database.endpoint_statistics_table import EndpointStatisticsTable
@@ -421,45 +430,6 @@ def test_process_region_request(
             image_id=TEST_IMAGE_ID, region_id="test-region-id", region_pixel_bounds="(0, 0)(50, 50)"
         )
 
-        region_queue_put_calls = [
-            mock.call(RegionRequestMatcher({"region": ((0, 0), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((0, 9), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((0, 18), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((0, 27), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((0, 36), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((0, 45), (5, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((9, 0), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((9, 9), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((9, 18), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((9, 27), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((9, 36), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((9, 45), (5, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((18, 0), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((18, 9), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((18, 18), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((18, 27), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((18, 36), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((18, 45), (5, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((27, 0), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((27, 9), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((27, 18), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((27, 27), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((27, 36), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((27, 45), (5, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((36, 0), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((36, 9), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((36, 18), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((36, 27), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((36, 36), (10, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((36, 45), (5, 10)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((45, 0), (10, 5)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((45, 9), (10, 5)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((45, 18), (10, 5)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((45, 27), (10, 5)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((45, 36), (10, 5)), "image_id": "test-image-id"})),
-            mock.call(RegionRequestMatcher({"region": ((45, 45), (5, 5)), "image_id": "test-image-id"})),
-        ]
-
         # Load up our test image
         raster_dataset, sensor_model = load_gdal_dataset(self.region_request.image_url)
 
@@ -467,27 +437,145 @@ def test_process_region_request(
         self.model_runner.region_request_table = Mock(RegionRequestTable, autospec=True)
         self.model_runner.endpoint_statistics_table = Mock(EndpointStatisticsTable, autospec=True)
         self.model_runner.endpoint_statistics_table.current_in_progress_regions.return_value = 0
+        self.model_runner.fail_region_request = Mock()
+
+        mock_setup_tile_workers.return_value = (Queue(), [])
+        mock_process_tiles.return_value = (10, 10)
+
         self.model_runner.process_region_request(self.region_request, region_request_item, raster_dataset, sensor_model)
 
-        # Create tile worker threads to process tiles
-        num_workers = int(os.environ["WORKERS"])
-        for i in range(num_workers):
-            region_queue_put_calls.append(mock.call(RegionRequestMatcher(None)))
+        mock_process_tiles.assert_called_once()
+        self.model_runner.fail_region_request.assert_not_called()
+
+        # We're testing a single region here so expecting a single call to both increment and
+        # decrement for the model associated with the region
+        self.model_runner.endpoint_statistics_table.increment_region_count.assert_called_once_with(TEST_MODEL_ENDPOINT)
+        self.model_runner.endpoint_statistics_table.decrement_region_count.assert_called_once_with(TEST_MODEL_ENDPOINT)
+
+    @mock.patch("aws.osml.model_runner.app.setup_tile_workers")
+    @mock.patch("aws.osml.model_runner.app.process_tiles")
+    def test_process_region_request_exception(
+        self,
+        mock_process_tiles,
+        mock_setup_tile_workers,
+    ):
+        from aws.osml.gdal.gdal_utils import load_gdal_dataset
+        from aws.osml.model_runner.database.endpoint_statistics_table import EndpointStatisticsTable
+        from aws.osml.model_runner.database.job_table import JobTable
+        from aws.osml.model_runner.database.region_request_table import RegionRequestItem, RegionRequestTable
+
+        region_request_item = RegionRequestItem(
+            image_id=TEST_IMAGE_ID, region_id="test-region-id", region_pixel_bounds="(0, 0)(50, 50)"
+        )
+
+        # Load up our test image
+        raster_dataset, sensor_model = load_gdal_dataset(self.region_request.image_url)
+
+        self.model_runner.job_table = Mock(JobTable, autospec=True)
+        self.model_runner.region_request_table = Mock(RegionRequestTable, autospec=True)
+        self.model_runner.endpoint_statistics_table = Mock(EndpointStatisticsTable, autospec=True)
+        self.model_runner.endpoint_statistics_table.current_in_progress_regions.return_value = 0
+        self.model_runner.fail_region_request = Mock()
+        mock_setup_tile_workers.return_value = (Queue(), [])
+        mock_process_tiles.side_effect = Exception("Mock processing exception")
+
+        self.model_runner.process_region_request(self.region_request, region_request_item, raster_dataset, sensor_model)
 
-        # Check to make sure the correct number of workers were created and setup with detectors and
-        # feature tables
-        assert mock_tile_worker.call_count == num_workers
-        assert mock_feature_detector.call_count == num_workers
-        assert mock_feature_table.call_count == num_workers
+        mock_process_tiles.assert_called_once()
+        self.model_runner.fail_region_request.assert_called_once()
 
         # We're testing a single region here so expecting a single call to both increment and
         # decrement for the model associated with the region
         self.model_runner.endpoint_statistics_table.increment_region_count.assert_called_once_with(TEST_MODEL_ENDPOINT)
         self.model_runner.endpoint_statistics_table.decrement_region_count.assert_called_once_with(TEST_MODEL_ENDPOINT)
 
+    @mock.patch("aws.osml.model_runner.app.setup_tile_workers")
+    @mock.patch("aws.osml.model_runner.app.process_tiles")
+    def test_process_region_request_invalid(
+        self,
+        mock_process_tiles,
+        mock_setup_tile_workers,
+    ):
+        from aws.osml.gdal.gdal_utils import load_gdal_dataset
+        from aws.osml.model_runner.api import RegionRequest
+        from aws.osml.model_runner.database.endpoint_statistics_table import EndpointStatisticsTable
+        from aws.osml.model_runner.database.job_table import JobTable
+        from aws.osml.model_runner.database.region_request_table import RegionRequestItem, RegionRequestTable
+
+        invalid_region_request = RegionRequest(
+            {
+                "tile_size": (10, 10),
+                "tile_overlap": (1, 1),
+                "tile_format": "NITF",
+                "image_id": TEST_IMAGE_ID,
+                "image_url": TEST_IMAGE_FILE,
+                "region_bounds": ((0, 0), (50, 50)),
+                "model_invoke_mode": "SM_ENDPOINT",
+                "image_extension": TEST_IMAGE_EXTENSION,
+            }
+        )
+
+        region_request_item = RegionRequestItem(
+            image_id=TEST_IMAGE_ID, region_id="test-region-id", region_pixel_bounds="(0, 0)(50, 50)"
+        )
+
+        # Load up our test image
+        raster_dataset, sensor_model = load_gdal_dataset(self.region_request.image_url)
+
+        self.model_runner.job_table = Mock(JobTable, autospec=True)
+        self.model_runner.region_request_table = Mock(RegionRequestTable, autospec=True)
+        self.model_runner.endpoint_statistics_table = Mock(EndpointStatisticsTable, autospec=True)
+        self.model_runner.endpoint_statistics_table.current_in_progress_regions.return_value = 0
+        mock_setup_tile_workers.return_value = (Queue(), [])
+        mock_process_tiles.return_value = (10, 10)
+
+        with self.assertRaises(ValueError):
+            self.model_runner.process_region_request(
+                invalid_region_request, region_request_item, raster_dataset, sensor_model
+            )
+
+        # We're testing a single region here so expecting a single call to both increment and
+        # decrement for the model associated with the region
+        self.model_runner.endpoint_statistics_table.increment_region_count.assert_not_called()
+        self.model_runner.endpoint_statistics_table.decrement_region_count.assert_not_called()
+
+    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.FeatureDetectorFactory", autospec=True)
+    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.FeatureTable", autospec=True)
+    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.TileWorker", autospec=True)
+    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.Queue", autospec=True)
+    def test_process_region_request_throttled(
+        self,
+        mock_queue,
+        mock_tile_worker,
+        mock_feature_table,
+        mock_feature_detector,
+    ):
+        from aws.osml.gdal.gdal_utils import load_gdal_dataset
+        from aws.osml.model_runner.database.endpoint_statistics_table import EndpointStatisticsTable
+        from aws.osml.model_runner.database.job_table import JobTable
+        from aws.osml.model_runner.database.region_request_table import RegionRequestTable
+        from aws.osml.model_runner.exceptions import SelfThrottledRegionException
+
+        # Load up our test image
+        raster_dataset, sensor_model = load_gdal_dataset(self.region_request.image_url)
+
+        self.model_runner.job_table = Mock(JobTable, autospec=True)
+        self.model_runner.region_request_table = Mock(RegionRequestTable, autospec=True)
+        self.model_runner.endpoint_statistics_table = Mock(EndpointStatisticsTable, autospec=True)
+        self.model_runner.endpoint_statistics_table.current_in_progress_regions.return_value = 10000
+
+        with self.assertRaises(SelfThrottledRegionException):
+            self.model_runner.process_region_request(self.region_request, raster_dataset, sensor_model)
+
+        self.model_runner.endpoint_statistics_table.increment_region_count.assert_not_called()
+        self.model_runner.endpoint_statistics_table.decrement_region_count.assert_not_called()
+
+        assert mock_tile_worker.call_count == 0
+        assert mock_feature_detector.call_count == 0
+        assert mock_feature_table.call_count == 0
+
         # Check to make sure a queue was created and populated with appropriate region requests
-        mock_queue.assert_called_once()
-        mock_queue.return_value.put.assert_has_calls(region_queue_put_calls)
+        mock_queue.assert_not_called()
 
     @mock.patch.dict("os.environ", values={"ELEVATION_DATA_LOCATION": TEST_ELEVATION_DATA_LOCATION})
     def test_create_elevation_model(self):
@@ -533,43 +621,29 @@ def test_create_elevation_model_disabled(self):
         elevation_model = ModelRunner.create_elevation_model()
         assert not elevation_model
 
-    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.FeatureDetectorFactory", autospec=True)
-    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.FeatureTable", autospec=True)
-    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.TileWorker", autospec=True)
-    @mock.patch("aws.osml.model_runner.tile_worker.tile_worker_utils.Queue", autospec=True)
-    def test_process_region_request_throttled(
-        self,
-        mock_queue,
-        mock_tile_worker,
-        mock_feature_table,
-        mock_feature_detector,
-    ):
-        from aws.osml.gdal.gdal_utils import load_gdal_dataset
-        from aws.osml.model_runner.database.endpoint_statistics_table import EndpointStatisticsTable
-        from aws.osml.model_runner.database.job_table import JobTable
-        from aws.osml.model_runner.database.region_request_table import RegionRequestTable
-        from aws.osml.model_runner.exceptions import SelfThrottledRegionException
-
-        # Load up our test image
-        raster_dataset, sensor_model = load_gdal_dataset(self.region_request.image_url)
+    def test_calculate_region_status_success(self):
+        from aws.osml.model_runner.common import RegionRequestStatus
 
-        self.model_runner.job_table = Mock(JobTable, autospec=True)
-        self.model_runner.region_request_table = Mock(RegionRequestTable, autospec=True)
-        self.model_runner.endpoint_statistics_table = Mock(EndpointStatisticsTable, autospec=True)
-        self.model_runner.endpoint_statistics_table.current_in_progress_regions.return_value = 10000
+        total_count = 10
+        error_count = 0
+        status = self.model_runner.calculate_region_status(total_count, error_count)
+        assert status == RegionRequestStatus.SUCCESS
 
-        with self.assertRaises(SelfThrottledRegionException):
-            self.model_runner.process_region_request(self.region_request, raster_dataset, sensor_model)
+    def test_calculate_region_status_partial(self):
+        from aws.osml.model_runner.common import RegionRequestStatus
 
-        self.model_runner.endpoint_statistics_table.increment_region_count.assert_not_called()
-        self.model_runner.endpoint_statistics_table.decrement_region_count.assert_not_called()
+        total_count = 10
+        error_count = 5
+        status = self.model_runner.calculate_region_status(total_count, error_count)
+        assert status == RegionRequestStatus.PARTIAL
 
-        assert mock_tile_worker.call_count == 0
-        assert mock_feature_detector.call_count == 0
-        assert mock_feature_table.call_count == 0
+    def test_calculate_region_status_failure(self):
+        from aws.osml.model_runner.common import RegionRequestStatus
 
-        # Check to make sure a queue was created and populated with appropriate region requests
-        mock_queue.assert_not_called()
+        total_count = 10
+        error_count = 10
+        status = self.model_runner.calculate_region_status(total_count, error_count)
+        assert status == RegionRequestStatus.FAILED
 
     @staticmethod
     def get_dataset_and_camera():