mark packages as failed when the state is upload failed

[Dert1129]

Summary by CodeRabbit

• Bug Fixes

  • Corrected database query syntax to ensure accurate package lookups.

• New Features

  • System now automatically detects packages that fail during upload and marks them for error handling. Improved error tracking with timezone-aware data handling.

[coderabbitai]

Walkthrough

This PR adds error handling to detect and mark packages with failed uploads. The changes introduce a workflow that scans MongoDB for failed uploads, retrieves corresponding packages, and marks them as failed in the inventory database. A SQL query bug fix is included. The new error-marking step runs once per iteration in the main watch loop.

Changes

Upload Failure Detection and Package Error Marking

Layer / File(s)	Summary
MongoDB state collection and failed upload queries data_management/services/dlu_mongo.py	Initialize state_collection with timezone-aware codec options and add find_by_upload_failed() method to query for documents with state = "UPLOAD_FAILED".
Package error marking in database data_management/services/dlu_management.py	Add set_dlu_package_error() method to update dlu_error = 1 in the package inventory, and fix a malformed SQL SELECT statement in get_package().
Watcher orchestration and event loop integration data_management/watch_files.py	Implement mark_packages_with_error() to scan for upload failures, look up packages, and mark them as failed if not already marked. Integrate the error-marking step into the main polling loop.

🚥 Pre-merge checks | ✅ 2

✅ Passed checks (2 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch KPMP-6657_change-dlu-error

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: aaa8dd70-619b-495a-ac89-6dfb34e46a7f

📥 Commits

Reviewing files that changed from the base of the PR and between e931c91 and 0172464.

📒 Files selected for processing (3)

• data_management/services/dlu_management.py
• data_management/services/dlu_mongo.py
• data_management/watch_files.py

[coderabbitai]

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Guard missing packageId before dereferencing failed state docs.

At Line 203, direct indexing (failed_upload["packageId"]) can raise KeyError and break this pass if any malformed state document is present. Add a defensive read and skip/log invalid rows.

Proposed fix

     def mark_packages_with_error(self):
         failed_uploads = self.dlu_mongo.find_by_upload_failed()
         for failed_upload in failed_uploads:
-            package_id = failed_upload["packageId"]
+            package_id = failed_upload.get("packageId")
+            if not package_id:
+                logger.warning("Skipping UPLOAD_FAILED record without packageId: %s", failed_upload.get("_id"))
+                continue
             package = self.dlu_management.get_package(package_id)
             if package is not None:
                 if package['dlu_error'] == 1:
                     continue
                 else:
-                    self.dlu_management.set_dlu_package_error(failed_upload["packageId"])
-                    logger.info("Marked package " + failed_upload["packageId"] + " as error due to UPLOAD_FAILED in MongoDB")
+                    self.dlu_management.set_dlu_package_error(package_id)
+                    logger.info("Marked package %s as error due to UPLOAD_FAILED in MongoDB", package_id)

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change

	package_id = failed_upload["packageId"]
	package = self.dlu_management.get_package(package_id)
	if package is not None:
	if package['dlu_error'] == 1:
	continue
	else:
	self.dlu_management.set_dlu_package_error(failed_upload["packageId"])
	logger.info("Marked package " + failed_upload["packageId"] + " as error due to UPLOAD_FAILED in MongoDB")
	package_id = failed_upload.get("packageId")
	if not package_id:
	logger.warning("Skipping UPLOAD_FAILED record without packageId: %s", failed_upload.get("_id"))
	continue
	package = self.dlu_management.get_package(package_id)
	if package is not None:
	if package['dlu_error'] == 1:
	continue
	else:
	self.dlu_management.set_dlu_package_error(package_id)
	logger.info("Marked package %s as error due to UPLOAD_FAILED in MongoDB", package_id)