jobflow admin upgrade fails - BSON error

[msiron-entalpic]

Command:
jf admin upgrade

Returns:

The selected project is AWS_DFT from config file /home/martin_siron/.jfremote/AWS_DFT.yaml
/home/martin_siron/software/entalcat/.venv/lib/python3.11/site-packages/maggma/stores/mongolike.py:501: UserWarning: You appear to be connected to a DocumentDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/documentdb
  conn: MongoClient = MongoClient(self.uri, **self.mongoclient_kwargs)
/home/martin_siron/software/entalcat/.venv/lib/python3.11/site-packages/maggma/stores/mongolike.py:111: UserWarning: You appear to be connected to a DocumentDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/documentdb
  MongoClient(
/home/martin_siron/software/entalcat/.venv/lib/python3.11/site-packages/maggma/stores/gridfs.py:147: UserWarning: You appear to be connected to a DocumentDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/documentdb
  MongoClient(
An Error occurred during the command execution: OperationFailure Failed to parse update: field must be of BSON type object, full error: {'ok': 0.0, 'code': 9, 'errmsg': 'Failed 
to parse update: field must be of BSON type object', 'operationTime': Timestamp(1747902153, 1)}

Debugging:
Using breakpoint I've identified that the error comes at the following line in utils/db.py

while True:
            result = self.collection.find_one_and_update(
                db_filter,
                update,
                upsert=False,
                return_document=ReturnDocument.AFTER,
                **self.kwargs,
            )

The issue appears that update is a list rather than a dictionary.
[{'$set': {'lock_id': {'$cond': {'if': {'$gt': ['$lock_id', None]}, 'then': '$lock_id', 'else': '415216cb-4eac-4153-8300-aa4bef591e29'}}, 'lock_time': {'$cond': {'if': {'$gt': ['$lock_id', None]}, 'then': '$lock_time', 'else': datetime.datetime(2025, 5, 22, 8, 24, 14, 987800)}}}}]

Indeed, changing update to update[0] in the collection.find_one_and_update() line appears to go through with no error, but leads to downstream errors when attempting to start the runner.

Details:
jobflow-remote version 0.1.6
maggma version: 0.71.5

[gpetretto]

Hi @msiron-entalpic, thanks for reporting this and for the detailed investigation.
Looking at your stack trace I see that there is this warning:

You appear to be connected to a DocumentDB cluster. For more information regarding feature compatibility and support please visit https://www.mongodb.com/supportability/documentdb

suggesting that you are using the AWS DocumentDB, so it might be something that usually works for standard MongoDB (or at least relatively recent versions).
In principle the list of dictionary is a valid input for collection.find_one_and_update (seed docs and code).
This part of the code was written quite some time ago, but if I remember correctly, giving it as a list was needed to make sure that the DB interprets it as a pipeline.
It might be that this functionality is not supported by DocumentDB.

Was everything working fine before the upgrade? From which version did you perform the upgrade?

It may also be useful to have the error that you get when you try to start the runner. In the CLI, if you want the full stack trace of the error you can use the -fe option. To be used just after the jf. For example jf -fe runner start.

[msiron-entalpic]

Hello! Sorry for the delay. In fact this is for the database reset, not the database upgrade. And it indeed seems to be related to the findAndModify feature, which is supported in MongoDB but not in DocumentsDB. This problem does not appear in MongoDB database.

The fix for a DocumentsDB would be to transfer the list to do the action one by one it seems.

I cannot post the -fe as it shows hosts, passwords, etc. I would have to clean it, but ultimately it fails on:

/home/martin_siron/software/entalcat/.venv/lib/python3.11/site-packages/pymongo/synchronous/network.py:218 in command                                                                │
│                                                                                                                                                                                      │
│   215 │   │   │   if client:                                                                                                                                                         │
│   216 │   │   │   │   client._process_response(response_doc, session)                                                                                                                │
│   217 │   │   │   if check:                                                                                                                                                          │
│ ❱ 218 │   │   │   │   helpers_shared._check_command_response(                                                                                                                        │
│   219 │   │   │   │   │   response_doc,                                                                                                                                              │
│   220 │   │   │   │   │   conn.max_wire_version,                                                                                                                                     │
│   221 │   │   │   │   │   allowable_errors,                                                                                                                                          │
│                                                                                                                                                                                      │
│ ╭───────────────────────────────────────────────────────────────────────────────────── locals ─────────────────────────────────────────────────────────────────────────────────────╮ │
│ │                   address = ('HOST', 27017)                                                                  │ │
│ │          allowable_errors = None                                                                                                                                                 │ │
│ │                     check = True                                                                                                                                                 │ │
│ │                    client = MongoClient(host=['HOST:27017'], document_class=dict, tz_aware=False,      │ │
│ │                             connect=True, tls=True, tlscafile='/home/martin_siron/global-bundle.pem', replicaset='rs0', readpreference='secondaryPreferred', retrywrites=False)  │ │
│ │             codec_options = CodecOptions(document_class=dict, tz_aware=False, uuid_representation=UuidRepresentation.UNSPECIFIED, unicode_decode_error_handler='strict',         │ │
│ │                             tzinfo=None, type_registry=TypeRegistry(type_codecs=[], fallback_encoder=None), datetime_conversion=DatetimeConversion.DATETIME)                     │ │
│ │                 collation = None                                                                                                                                                 │ │
│ │           compression_ctx = None                                                                                                                                                 │ │
│ │                      conn = Connection(<ssl.SSLSocket fd=7, family=2, type=1, proto=6, laddr=('10.98.128.208', 39930), raddr=('10.98.128.226', 27017)>) at 140569116179088       │ │
│ │                    dbname = 'jobflow'                                                                                                                                            │ │
│ │                  duration = datetime.timedelta(microseconds=208)                                                                                                                 │ │
│ │           exhaust_allowed = False                                                                                                                                                │ │
│ │                   failure = {'ok': 0.0, 'code': 9, 'errmsg': 'Failed to parse update: field must be of BSON type object', 'operationTime': Timestamp(1749132011, 1)}             │ │
│ │                     flags = 0                                                                                                                                                    │ │
│ │                 is_mongos = False                                                                                                                                                │ │
│ │                 listeners = <pymongo.monitoring._EventListeners object at 0x7fd8cd105b90>                                                                                        │ │
│ │             max_bson_size = 16777216                                                                                                                                             │ │
│ │              max_doc_size = 0                                                                                                                                                    │ │
│ │                       msg = b'\xb0\x01\x00\x00\xcd|\xe8F\x00\x00\x00\x00\xdd\x07\x00\x00\x00\x00\x00\x00\x00\x9b\x01\x00\x00\x02findAndModify\x00\r\x00\x00\x00jf_auxiliary\x00… │ │
│ │                      name = 'findAndModify'                                                                                                                                      │ │
│ │                        ns = 'jobflow.$cmd'                                                                                                                                       │ │
│ │                      orig = {                                                                                                                                                    │ │
│ │                             │   'findAndModify': 'jf_auxiliary',                                                                                                                 │ │
│ │                             │   'query': {'running_runner': {'$exists': True}},                                                                                                  │ │
│ │                             │   'new': True,                                                                                                                                     │ │
│ │                             │   'update': [                                                                                                                                      │ │
│ │                             │   │   {                                                                                                                                            │ │
│ │                             │   │   │   '$set': {                                                                                                                                │ │
│ │                             │   │   │   │   'lock_id': {'$cond': {'if': {'$gt': ['$lock_id', None]}, 'then': '$lock_id', 'else': '3f020aa5-9476-4d7e-9b81-2e2360acdbcc'}},       │ │
│ │                             │   │   │   │   'lock_time': {                                                                                                                       │ │
│ │                             │   │   │   │   │   '$cond': {'if': {'$gt': ['$lock_id', None]}, 'then': '$lock_time', 'else': datetime.datetime(2025, 6, 5, 14, 0, 11, 702460)}     │ │
│ │                             │   │   │   │   }                                                                                                                                    │ │
│ │                             │   │   │   }                                                                                                                                        │ │
│ │                             │   │   }                                                                                                                                            │ │
│ │                             │   ],                                                                                                                                               │ │
│ │                             │   'upsert': False,                                                                                                                                 │ │
│ │                             │   'lsid': {'id': Binary(b'\x95.\x10^\xa7<J\xc0\xa6\xf6\x01\xc2\xa4\xe8p\x03', 4)},                                                                 │ │
│ │                             │   '$db': 'jobflow'                                                                                                                                 │ │
│ │                             }                                                                                                                                                    │ │
│ │ parse_write_concern_error = True                                                                                                                                                 │ │
│ │                   publish = False                                                                                                                                                │ │
│ │              read_concern = None                                                                                                                                                 │ │
│ │           read_preference = Primary()                                                                                                                                            │ │
│ │                     reply = <pymongo.message._OpMsg object at 0x7fd8cc3c6180>                                                                                                    │ │
│ │                request_id = 1189641421                                                                                                                                           │ │
│ │              response_doc = {'ok': 0.0, 'code': 9, 'errmsg': 'Failed to parse update: field must be of BSON type object', 'operationTime': Timestamp(1749132011, 1)}             │ │
│ │                   session = <pymongo.synchronous.client_session.ClientSession object at 0x7fd8cd106dd0>                                                                          │ │
│ │                      size = 411                                                                                                                                                  │ │
│ │                      spec = {                                                                                                                                                    │ │
│ │                             │   'findAndModify': 'jf_auxiliary',                                                                                                                 │ │
│ │                             │   'query': {'running_runner': {'$exists': True}},                                                                                                  │ │
│ │                             │   'new': True,                                                                                                                                     │ │
│ │                             │   'update': [                                                                                                                                      │ │
│ │                             │   │   {                                                                                                                                            │ │
│ │                             │   │   │   '$set': {                                                                                                                                │ │
│ │                             │   │   │   │   'lock_id': {'$cond': {'if': {'$gt': ['$lock_id', None]}, 'then': '$lock_id', 'else': '3f020aa5-9476-4d7e-9b81-2e2360acdbcc'}},       │ │
│ │                             │   │   │   │   'lock_time': {                                                                                                                       │ │
│ │                             │   │   │   │   │   '$cond': {'if': {'$gt': ['$lock_id', None]}, 'then': '$lock_time', 'else': datetime.datetime(2025, 6, 5, 14, 0, 11, 702460)}     │ │
│ │                             │   │   │   │   }                                                                                                                                    │ │
│ │                             │   │   │   }                                                                                                                                        │ │
│ │                             │   │   }                                                                                                                                            │ │
│ │                             │   ],                                                                                                                                               │ │
│ │                             │   'upsert': False,                                                                                                                                 │ │
│ │                             │   'lsid': {'id': Binary(b'\x95.\x10^\xa7<J\xc0\xa6\xf6\x01\xc2\xa4\xe8p\x03', 4)},                                                                 │ │
│ │                             │   '$db': 'jobflow'                                                                                                                                 │ │
│ │                             }                                                                                                                                                    │ │
│ │         speculative_hello = False                                                                                                                                                │ │
│ │                     start = datetime.datetime(2025, 6, 5, 14, 0, 11, 712386)                                                                                                     │ │
│ │            unacknowledged = False                                                                                                                                                │ │
│ │             unpacked_docs = [{'ok': 0.0, 'code': 9, 'errmsg': 'Failed to parse update: field must be of BSON type object', 'operationTime': Timestamp(1749132011, 1)}]           │ │


│ │                use_op_msg = True                                                                                                                                                 │ │
│ │               user_fields = {'value': 1}                                                                                                                                         │ │
│ │             write_concern = WriteConcern()                                                                                                                                       │ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ │
│                                                                                                                                                                                      │
│ /home/martin_siron/software/entalcat/.venv/lib/python3.11/site-packages/pymongo/helpers_shared.py:247 in _check_command_response                                                     │
│                                                                                                                                                                                      │
│   244 │   elif code == 43:                                                                                                                                                           │
│   245 │   │   raise CursorNotFound(errmsg, code, response, max_wire_version)                                                                                                         │
│   246 │                                                                                                                                                                              │
│ ❱ 247 │   raise OperationFailure(errmsg, code, response, max_wire_version)                                                                                                           │
│   248                                                                                                                                                                                │
│   249                                                                                                                                                                                │
│   250 def _raise_last_write_error(write_errors: list[Any]) -> NoReturn:                                                                                                              │
│                                                                                                                                                                                      │
│ ╭─────────────────────────────────────────────────────────────────────────────── locals ───────────────────────────────────────────────────────────────────────────────╮             │
│ │          allowable_errors = None                                                                                                                                     │             │
│ │                      code = 9                                                                                                                                        │             │
│ │                   details = {'ok': 0.0, 'code': 9, 'errmsg': 'Failed to parse update: field must be of BSON type object', 'operationTime': Timestamp(1749132011, 1)} │             │
│ │                    errmsg = 'Failed to parse update: field must be of BSON type object'                                                                              │             │
│ │          max_wire_version = 13                                                                                                                                       │             │
│ │ parse_write_concern_error = True                                                                                                                                     │             │
│ │                  response = {'ok': 0.0, 'code': 9, 'errmsg': 'Failed to parse update: field must be of BSON type object', 'operationTime': Timestamp(1749132011, 1)} │  

using pymongo I was able to test the pipeline vs. individual:
This example below works:

existing_document = collection.find_one({"running_runner": {"$exists": True}})

if existing_document and 'lock_id' in existing_document:
    lock_id = existing_document['lock_id']
    lock_time = existing_document['lock_time']
else:
    lock_id = '3f020aa5-9476-4d7e-9b81-2e2360acdbcc'
    lock_time = datetime.datetime(2025, 6, 5, 14, 0, 11, 702460)

collection.find_one_and_update(
    {"running_runner": {"$exists": True}},
    {"$set": {"lock_id": lock_id, "lock_time": lock_time}},
    upsert=False,
    return_document=True
)

While this does not:

collection.find_one_and_update(
    {"running_runner": {"$exists": True}},
    [
        {
            "$set": {
                "lock_id": {
                    "$cond": [
                        {"$gt": ["$lock_id", None]},
                        "$lock_id",
                        "3f020aa5-9476-4d7e-9b81-2e2360acdbcc"
                    ]
                },
                "lock_time": {
                    "$cond": [
                        {"$gt": ["$lock_id", None]},
                        "$lock_time",
                        datetime.datetime(2025, 6, 5, 14, 0, 11, 702460)
                    ]
                }
            }
        }
    ],
    upsert=False,
    return_document=True
)

Returning the same error:

OperationFailure: Failed to parse update: field must be of BSON type object, full error: {'ok': 0.0, 'code': 9, 'errmsg': 'Failed to parse update: field must be of BSON type object', 'operationTime': Timestamp(1749133890, 1)}

[gpetretto]

Hi, thanks for providing additional details.
Unfortunately, switching to a version where the find is split from the update of the document will not be possible. In that case the operation will not be atomic anymore and race conditions might lead to multiple processes acquiring the lock on the same job.

I don't know if it would be feasible to create an alternative version of that portion of the code that could be chosen based on a configuration option. It depends how many points of the code need to modified and maintained. In any case one would need to live with the risk of failures due to simulatanous locks.
Another minor issue is that I did not find a github action for the AWS DocumentDB, so we would not be able to check in the CI tests that compatibility is preserved for future updates.

I doubt I will have the time to implement such an option, but if the changes do not impact too much of the code I can review a PR.