YDA-6753: add support for async checksum calculation

__init__.py

@@ -32,6 +32,7 @@
 from admin                    import *
 from arb                      import *
 from browse                   import *
+from checksums                import *
 from datacite                 import *
 from folder                   import *
 from groups                   import *

checksums.py

@@ -0,0 +1,30 @@
+"""Functions for data object checksums."""
+
+__copyright__ = 'Copyright (c) 2026, Utrecht University'
+__license__   = 'GPLv3, see LICENSE'
+
+import irods_types
+
+from util import data_object, log, msi, rule
+
+__all__ = ['rule_verify_checksum']
+
+
+@rule.make()
+def rule_verify_checksum(ctx: rule.Context, path: str) -> None:
+    """Verify checksum of a data object, calculate checksum if it doesn't exist.
+
+    :param ctx:  Combined type of a callback and rei struct
+    :param path: Path of data object to verify
+    """
+    if data_object.checksum(ctx, path):
+        options = "verifyChksum="
+    else:
+        options = "ChksumAll=++++forceChksum="
+
+    try:
+        msi.data_obj_chksum(ctx, path, options, irods_types.BytesBuf())
+        checksum = data_object.checksum(ctx, path)
+        log.write(ctx, f"Verified checksum of <{path}>: {checksum}")
+    except Exception as e:
+        log.write(ctx, f"Could not verify checksum of <{path}>: {str(e)}")

policies.py

@@ -617,9 +617,15 @@ def pep_resource_modified_post(ctx: rule.Context,
         # Log errors, but continue with revisions.
         log.write(ctx, 'rule_meta_modified_post failed: ' + str(e))
 
-    # ctx.uuResourceModifiedPostRevision(instance_name, zone, path)
     revisions.resource_modified_post_revision(ctx, instance_name, zone, path)
 
+    if config.enable_async_checksum:
+        # Calculate and verify checksum of data object.
+        ctx.delayExec(
+            f"<PLUSET>{config.async_checksum_delay_time}s</PLUSET><INST_NAME>irods_rule_engine_plugin-irods_rule_language-instance</INST_NAME>",
+            f"rule_verify_checksum('{path}')",
+            "")
+
 
 @rule.make()
 def py_acPostProcForObjRename(ctx: rule.Context, src: str, dst: str) -> None:

rules_uu.cfg.template

@@ -53,6 +53,9 @@ token_lifetime                 =
 enable_inactivity_notification =
 inactivity_cutoff_months       =
 
+enable_async_checksum          =
+async_checksum_delay_time      =
+
 async_replication_delay_time   =
 async_replication_max_rss      =
 async_revision_delay_time      =

util/config.py

@@ -117,6 +117,8 @@ def __repr__(self) -> str:
                 token_length=0,
                 token_lifetime=0,
                 token_expiration_notification=0,
+                enable_async_checksum=False,
+                async_checksum_delay_time=0,
                 async_replication_delay_time=0,
                 async_replication_max_rss=1000000000,
                 async_revision_delay_time=0,

util/data_object.py

@@ -101,6 +101,29 @@ def size(ctx: rule.Context, path: str) -> int | None:
     return None
 
 
+def checksum(ctx: rule.Context, path: str) -> str | None:
+    """Get a data object's checksum.
+
+    :param ctx:  Combined type of a callback and rei struct
+    :param path: Path to iRODS data object
+
+    :returns: Data object's checksum or None if object is not found
+    """
+    coll_name, data_name = pathutil.chop(path)
+    coll_name = misc.escape(coll_name)
+    data_name = misc.escape(data_name)
+    iter = genquery.Query(
+        ctx, "DATA_CHECKSUM",
+        f"COLL_NAME = '{coll_name}' AND DATA_NAME = '{data_name}'",
+        output=genquery.AS_LIST
+    )
+
+    for row in iter:
+        return row[0]
+
+    return None
+
+
 def has_replica_with_status(ctx: rule.Context, path: str, statuses: List) -> bool:
     """Check if data object has replica with specified replica statuses.