From 032fdad597f1ded3abef346ee7ac553cc9efd064 Mon Sep 17 00:00:00 2001 From: wangyinz Date: Wed, 12 Aug 2026 18:06:10 -0500 Subject: [PATCH 1/5] Make GridFS waveform replacement failure-safe --- python/mspasspy/db/database.py | 161 +++++++----- .../db/test_gridfs_overwrite_contract.py | 246 ++++++++++++++++++ 2 files changed, 344 insertions(+), 63 deletions(-) create mode 100644 python/tests/db/test_gridfs_overwrite_contract.py diff --git a/python/mspasspy/db/database.py b/python/mspasspy/db/database.py index 1459505b41..3b72c3286e 100644 --- a/python/mspasspy/db/database.py +++ b/python/mspasspy/db/database.py @@ -3169,6 +3169,11 @@ def update_data( # only for usage errors. We test the elog size to check if there # are other warning messages and add a summary if there were any logsize0 = mspass_object.elog.size() + old_gridfs_id = ( + mspass_object["gridfs_id"] if "gridfs_id" in mspass_object else None + ) + if old_gridfs_id is not None: + mspass_object.erase("gridfs_id") try: self.update_metadata( mspass_object, @@ -3179,8 +3184,9 @@ def update_data( normalizing_collections=normalizing_collections, alg_name=alg_name, ) - except: - raise + finally: + if old_gridfs_id is not None: + mspass_object["gridfs_id"] = old_gridfs_id logsize = mspass_object.elog.size() # A bit verbose, but we post this warning to make it clear the # problem originated from update_data - probably not really needed @@ -3227,69 +3233,90 @@ def update_data( ) mspass_object["storage_mode"] = "gridfs" update_record["storage_mode"] = "gridfs" - # This logic overwrites the content if the magic key - # "gridfs_id" exists in the input. In both cases gridfs_id is set - # in returned - if "gridfs_id" in mspass_object: + # Supplying the replacement id lets rollback remove a blob even + # when GridFS writes it and then raises an exception. + staged_gridfs_id = ObjectId() if old_gridfs_id is not None else None + try: mspass_object = self._save_sample_data_to_gridfs( mspass_object, - overwrite=True, - ) - else: - mspass_object = self._save_sample_data_to_gridfs( - mspass_object, overwrite=False + overwrite=False, + gridfs_id=staged_gridfs_id, ) + new_gridfs_id = mspass_object["gridfs_id"] + update_record["gridfs_id"] = new_gridfs_id - # There is a possible efficiency gain right here. Not sure if - # gridfs_id is altered when the sample data are updated in place. - # if we can be sure the returned gridfs_id is the same as theresult - # input in that case, we would omit gridfs_id from the update - # record and most data would not require the final update - # transaction below - update_record["gridfs_id"] = mspass_object["gridfs_id"] - # should define wf_collection here because if the mspass_object is dead - if collection: - wf_collection_name = collection - else: - # This returns a string that is the collection name for this atomic data type - # A weird construct - wf_collection_name = save_schema.collection("_id") - wf_collection = self[wf_collection_name] - - elog_id = None - if mspass_object.elog.size() > 0: - elog_id_name = self.database_schema.default_name("elog") + "_id" - # FIXME I think here we should check if elog_id field exists in the mspass_object - # and we should update the elog entry if mspass_object already had one - if elog_id_name in mspass_object: - old_elog_id = mspass_object[elog_id_name] + # should define wf_collection here because if the mspass_object is dead + if collection: + wf_collection_name = collection else: - old_elog_id = None - # elog ids will be updated in the wf col when saving metadata - elog_id = self._save_elog( - mspass_object, elog_id=old_elog_id, data_tag=data_tag - ) - update_record[elog_id_name] = elog_id - - # update elog collection - # we have to do the xref to wf collection like this too - elog_col = self[self.database_schema.default_name("elog")] - wf_id_name = wf_collection_name + "_id" - filter_ = {"_id": elog_id} - elog_col.update_one( - filter_, {"$set": {wf_id_name: mspass_object["_id"]}} - ) - # finally we need to update the wf document if we set anything - # in update_record - if len(update_record): + # This returns a string that is the collection name for this atomic data type + # A weird construct + wf_collection_name = save_schema.collection("_id") + wf_collection = self[wf_collection_name] + + elog_id = None + if mspass_object.elog.size() > 0: + elog_id_name = self.database_schema.default_name("elog") + "_id" + # FIXME I think here we should check if elog_id field exists in the mspass_object + # and we should update the elog entry if mspass_object already had one + if elog_id_name in mspass_object: + old_elog_id = mspass_object[elog_id_name] + else: + old_elog_id = None + # elog ids will be updated in the wf col when saving metadata + elog_id = self._save_elog( + mspass_object, elog_id=old_elog_id, data_tag=data_tag + ) + update_record[elog_id_name] = elog_id + + # update elog collection + # we have to do the xref to wf collection like this too + elog_col = self[self.database_schema.default_name("elog")] + wf_id_name = wf_collection_name + "_id" + filter_ = {"_id": elog_id} + elog_col.update_one( + filter_, {"$set": {wf_id_name: mspass_object["_id"]}} + ) + filter_ = {"_id": mspass_object["_id"]} - wf_collection.update_one(filter_, {"$set": update_record}) - # we may probably set the elog_id field in the mspass_object - if elog_id: - mspass_object[elog_id_name] = elog_id - # we may probably set the history_object_id field in the mspass_object - if history_object_id: - mspass_object[history_obj_id_name] = history_object_id + if old_gridfs_id is not None: + filter_["gridfs_id"] = old_gridfs_id + result = wf_collection.update_one(filter_, {"$set": update_record}) + if old_gridfs_id is not None and result.matched_count != 1: + raise MsPASSError( + "Database.update_data could not replace the expected " + "GridFS reference", + ErrorSeverity.Invalid, + ) + except BaseException as original_error: + if old_gridfs_id is not None: + try: + gfsh = gridfs.GridFS(self) + if gfsh.exists(staged_gridfs_id): + gfsh.delete(staged_gridfs_id) + except Exception as cleanup_error: + raise original_error from cleanup_error + finally: + mspass_object["gridfs_id"] = old_gridfs_id + raise + # we may probably set the elog_id field in the mspass_object + if elog_id: + mspass_object[elog_id_name] = elog_id + # we may probably set the history_object_id field in the mspass_object + if history_object_id: + mspass_object[history_obj_id_name] = history_object_id + if old_gridfs_id is not None: + try: + gfsh = gridfs.GridFS(self) + if gfsh.exists(old_gridfs_id): + gfsh.delete(old_gridfs_id) + except Exception as error: + mspass_object.elog.log_error( + alg_name, + "GridFS replacement was committed, but the previous " + f"sample object could not be deleted: {error}", + ErrorSeverity.Complaint, + ) else: # Dead data land here elog_id_name = self.database_schema.default_name("elog") + "_id" @@ -6750,6 +6777,7 @@ def _save_sample_data_to_gridfs( self, mspass_object, overwrite=False, + gridfs_id=None, ): """ Saves the sample data array for a mspass seismic data object using @@ -6776,6 +6804,10 @@ def _save_sample_data_to_gridfs( set of documents will be created to hold the data in the gridfs system. Default is False. + :param gridfs_id: optional id to assign to the newly written GridFS + object. Used by update_data so a failed staged write can be rolled + back even if GridFS raises after creating the object. + :return: edited version of input (mspass_object). Return changed only by adding metadata attributes "storage_mode" and "gridfs_id" """ @@ -6785,12 +6817,15 @@ def _save_sample_data_to_gridfs( if isinstance(mspass_object, (TimeSeries, Seismogram)): if overwrite and mspass_object.is_defined("gridfs_id"): - gridfs_id = mspass_object["gridfs_id"] - if gfsh.exists(gridfs_id): - gfsh.delete(gridfs_id) + old_gridfs_id = mspass_object["gridfs_id"] + if gfsh.exists(old_gridfs_id): + gfsh.delete(old_gridfs_id) # verdion 2 had a transpose here that seemed unncessary ub = bytes(np.array(mspass_object.data)) - gridfs_id = gfsh.put(ub) + if gridfs_id is None: + gridfs_id = gfsh.put(ub) + else: + gridfs_id = gfsh.put(ub, _id=gridfs_id) mspass_object["gridfs_id"] = gridfs_id mspass_object["storage_mode"] = "gridfs" elif isinstance(mspass_object, (TimeSeriesEnsemble, SeismogramEnsemble)): diff --git a/python/tests/db/test_gridfs_overwrite_contract.py b/python/tests/db/test_gridfs_overwrite_contract.py new file mode 100644 index 0000000000..2b4e2455a6 --- /dev/null +++ b/python/tests/db/test_gridfs_overwrite_contract.py @@ -0,0 +1,246 @@ +import os +import uuid +from types import SimpleNamespace +from unittest.mock import patch + +import gridfs +import numpy as np +import pytest + +from mspasspy.ccore.seismic import ( + DoubleVector, + Seismogram, + TimeReferenceType, + TimeSeries, +) +from mspasspy.ccore.utility import AtomicType, ErrorSeverity, MsPASSError, dmatrix +from mspasspy.db.client import DBClient +from mspasspy.db.collection import Collection +from mspasspy.db.database import Database + + +@pytest.fixture +def database(): + uri = os.environ.get("MSPASS_TEST_MONGODB_URI", "mongodb://127.0.0.1:27017") + client = DBClient(uri, serverSelectionTimeoutMS=2000) + try: + client.admin.command("ping") + except Exception as error: + client.close() + pytest.skip(f"MongoDB is unavailable at {uri}: {error}") + name = "test_gridfs_overwrite_" + uuid.uuid4().hex + database = Database(client, name) + try: + yield database + finally: + client.drop_database(name) + client.close() + + +def make_timeseries(values): + datum = TimeSeries(len(values)) + datum.data = DoubleVector(values) + datum.set_as_origin("test", "0", "0", AtomicType.TIMESERIES) + return finish_datum(datum, len(values)) + + +def make_seismogram(values): + datum = Seismogram(len(values)) + samples = dmatrix(3, len(values)) + for component in range(3): + for sample, value in enumerate(values): + samples[component, sample] = value + 10.0 * component + datum.data = samples + datum.set_as_origin("test", "0", "0", AtomicType.SEISMOGRAM) + return finish_datum(datum, len(values)) + + +def finish_datum(datum, npts): + datum.dt = 0.1 + datum.t0 = 10.0 + datum.tref = TimeReferenceType.UTC + datum.set_live() + datum["npts"] = npts + datum["sampling_rate"] = 10.0 + datum["delta"] = 0.1 + datum["calib"] = 1.0 + return datum + + +def samples(datum): + return np.asarray(datum.data) + + +def save_original(database, factory): + return database.save_data( + factory([1.0, 2.0, 3.0, 4.0]), + mode="promiscuous", + storage_mode="gridfs", + save_history=False, + return_data=True, + ) + + +CASES = [ + (make_timeseries, "wf_TimeSeries"), + (make_seismogram, "wf_Seismogram"), +] + + +@pytest.mark.parametrize("factory,collection", CASES) +def test_successful_overwrite_commits_new_reference_before_removing_old_blob( + database, factory, collection +): + datum = save_original(database, factory) + old_gridfs_id = datum["gridfs_id"] + replacement = factory([5.0, 6.0, 7.0, 8.0]) + datum.data = replacement.data + + result = database.update_data(datum, mode="promiscuous") + + document = database[collection].find_one({"_id": datum["_id"]}) + new_gridfs_id = document["gridfs_id"] + assert result is datum + assert new_gridfs_id == datum["gridfs_id"] + assert new_gridfs_id != old_gridfs_id + storage = gridfs.GridFS(database) + assert not storage.exists(old_gridfs_id) + assert storage.exists(new_gridfs_id) + reread = database.read_data(document, collection=collection) + assert reread.live + np.testing.assert_allclose(samples(reread), samples(replacement)) + + +@pytest.mark.parametrize("factory,collection", CASES) +def test_new_put_failure_preserves_old_reference_and_blob( + database, factory, collection +): + datum = save_original(database, factory) + old_gridfs_id = datum["gridfs_id"] + old_document = database[collection].find_one({"_id": datum["_id"]}) + old_samples = samples(datum).copy() + datum.data = factory([5.0, 6.0, 7.0, 8.0]).data + failure = RuntimeError("injected GridFS put failure") + original_put = gridfs.GridFS.put + + def put_then_fail(self, data, *args, **kwargs): + original_put(self, data, *args, **kwargs) + raise failure + + with patch.object(gridfs.GridFS, "put", put_then_fail): + with pytest.raises(RuntimeError) as error: + database.update_data(datum, mode="promiscuous") + + assert error.value is failure + assert datum["gridfs_id"] == old_gridfs_id + document = database[collection].find_one({"_id": datum["_id"]}) + assert document["gridfs_id"] == old_gridfs_id + assert database["fs.files"].count_documents({}) == 1 + assert gridfs.GridFS(database).exists(old_gridfs_id) + reread = database.read_data(old_document, collection=collection) + assert reread.live + np.testing.assert_allclose(samples(reread), old_samples) + + +@pytest.mark.parametrize("factory,collection", CASES) +def test_reference_update_failure_removes_new_blob_and_rethrows_original_error( + database, factory, collection +): + datum = save_original(database, factory) + old_gridfs_id = datum["gridfs_id"] + old_document = database[collection].find_one({"_id": datum["_id"]}) + old_samples = samples(datum).copy() + datum.data = factory([5.0, 6.0, 7.0, 8.0]).data + failure = RuntimeError("injected waveform reference failure") + original_update_one = Collection.update_one + + def fail_reference_update(self, query, update, *args, **kwargs): + gridfs_id = update.get("$set", {}).get("gridfs_id") + if gridfs_id is not None and gridfs_id != old_gridfs_id: + raise failure + return original_update_one(self, query, update, *args, **kwargs) + + with patch.object(Collection, "update_one", fail_reference_update): + with pytest.raises(RuntimeError) as error: + database.update_data(datum, mode="promiscuous") + + assert error.value is failure + assert datum["gridfs_id"] == old_gridfs_id + document = database[collection].find_one({"_id": datum["_id"]}) + assert document["gridfs_id"] == old_gridfs_id + assert database["fs.files"].count_documents({}) == 1 + assert gridfs.GridFS(database).exists(old_gridfs_id) + reread = database.read_data(old_document, collection=collection) + assert reread.live + np.testing.assert_allclose(samples(reread), old_samples) + + +@pytest.mark.parametrize("factory,collection", CASES) +def test_reference_compare_miss_removes_new_blob_without_overwriting_current_reference( + database, factory, collection +): + datum = save_original(database, factory) + waveform_id = datum["_id"] + old_gridfs_id = datum["gridfs_id"] + datum.data = factory([5.0, 6.0, 7.0, 8.0]).data + final_queries = [] + original_update_one = Collection.update_one + + def miss_reference_update(self, query, update, *args, **kwargs): + gridfs_id = update.get("$set", {}).get("gridfs_id") + if gridfs_id is not None: + assert gridfs_id != old_gridfs_id + final_queries.append(query) + return SimpleNamespace(matched_count=0) + return original_update_one(self, query, update, *args, **kwargs) + + with patch.object(Collection, "update_one", miss_reference_update): + with pytest.raises(MsPASSError) as error: + database.update_data(datum, mode="promiscuous") + + assert error.value.severity == ErrorSeverity.Invalid + assert final_queries == [{"_id": waveform_id, "gridfs_id": old_gridfs_id}] + assert datum["gridfs_id"] == old_gridfs_id + document = database[collection].find_one({"_id": waveform_id}) + assert document["gridfs_id"] == old_gridfs_id + assert database["fs.files"].count_documents({}) == 1 + assert gridfs.GridFS(database).exists(old_gridfs_id) + + +@pytest.mark.parametrize("factory,collection", CASES) +def test_old_delete_failure_keeps_committed_new_data_and_logs_one_complaint( + database, factory, collection +): + datum = save_original(database, factory) + old_gridfs_id = datum["gridfs_id"] + replacement = factory([5.0, 6.0, 7.0, 8.0]) + datum.data = replacement.data + old_log_size = datum.elog.size() + failure = RuntimeError("injected old GridFS delete failure") + original_delete = gridfs.GridFS.delete + + def fail_old_delete(self, gridfs_id, *args, **kwargs): + if gridfs_id == old_gridfs_id: + raise failure + return original_delete(self, gridfs_id, *args, **kwargs) + + with patch.object(gridfs.GridFS, "delete", fail_old_delete): + result = database.update_data(datum, mode="promiscuous") + + document = database[collection].find_one({"_id": datum["_id"]}) + new_gridfs_id = document["gridfs_id"] + assert result is datum + assert new_gridfs_id == datum["gridfs_id"] + assert new_gridfs_id != old_gridfs_id + storage = gridfs.GridFS(database) + assert storage.exists(old_gridfs_id) + assert storage.exists(new_gridfs_id) + assert database["fs.files"].count_documents({}) == 2 + assert datum.elog.size() == old_log_size + 1 + complaint = datum.elog.get_error_log()[-1] + assert complaint.badness == ErrorSeverity.Complaint + assert "previous sample object could not be deleted" in complaint.message + assert str(failure) in complaint.message + reread = database.read_data(document, collection=collection) + assert reread.live + np.testing.assert_allclose(samples(reread), samples(replacement)) From 4f3225d78b5ba50c901c6fbb6dc2ec8710417560 Mon Sep 17 00:00:00 2001 From: wangyinz Date: Fri, 14 Aug 2026 03:03:36 -0500 Subject: [PATCH 2/5] Protect public GridFS overwrite paths --- python/mspasspy/db/database.py | 120 +++++++++++++++--- python/tests/db/test_database.py | 9 +- .../db/test_gridfs_overwrite_contract.py | 93 ++++++++++++-- 3 files changed, 192 insertions(+), 30 deletions(-) diff --git a/python/mspasspy/db/database.py b/python/mspasspy/db/database.py index 3b72c3286e..86aec5da45 100644 --- a/python/mspasspy/db/database.py +++ b/python/mspasspy/db/database.py @@ -1472,7 +1472,12 @@ def save_data( A discussion of format caveats can be found above. :type format: :class:`str` :param overwrite: If true gridfs data linked to the original - waveform will be replaced by the sample data from this save. + waveform will be replaced by the sample data from this save. A + previously saved atomic datum must carry both its waveform ``_id`` + and ``gridfs_id``; the existing waveform document is updated rather + than inserting a second document. Ensemble members follow the same + rule independently. Objects carrying neither reference are new + data and are saved normally. Default is false, and should be the normal use. This option should never be used after a reduce operator as the parents are not tracked and the space advantage is likely minimal for @@ -1554,7 +1559,80 @@ def save_data( ) raise MsPASSError(message, "Fatal") - if mspass_object.live: + overwrite_handled = False + if overwrite and storage_mode == "gridfs" and mspass_object.live: + if isinstance(mspass_object, (TimeSeries, Seismogram)): + has_wfid = "_id" in mspass_object + has_gridfs_id = "gridfs_id" in mspass_object + if has_wfid != has_gridfs_id: + raise ValueError( + "Database.save_data: overwrite=True requires both _id " + "and gridfs_id when either reference is defined" + ) + if has_wfid: + mspass_object = self.update_data( + mspass_object, + collection=collection, + mode=mode, + exclude_keys=exclude_keys, + data_tag=data_tag, + save_history=save_history, + normalizing_collections=normalizing_collections, + alg_name=alg_name, + alg_id=alg_id, + ) + overwrite_handled = True + else: + live_members = [d for d in mspass_object.member if d.live] + for d in live_members: + has_wfid = "_id" in d + has_gridfs_id = "gridfs_id" in d + if has_wfid != has_gridfs_id: + raise ValueError( + "Database.save_data: overwrite=True requires both " + "_id and gridfs_id for every previously saved member" + ) + if any("_id" in d for d in live_members): + mspass_object, bodies = self.stedronsky.bring_out_your_dead( + mspass_object + ) + if not cremate and len(bodies.member) > 0: + self.stedronsky.bury(bodies) + mspass_object.sync_metadata() + for d in mspass_object.member: + if d.live: + if "_id" in d: + d["storage_mode"] = "gridfs" + self.update_data( + d, + collection=collection, + mode=mode, + exclude_keys=exclude_keys, + data_tag=data_tag, + save_history=save_history, + normalizing_collections=normalizing_collections, + alg_name=alg_name, + alg_id=alg_id, + ) + else: + self.save_data( + d, + return_data=True, + mode=mode, + storage_mode="gridfs", + overwrite=False, + exclude_keys=exclude_keys, + collection=collection, + data_tag=data_tag, + cremate=cremate, + save_history=save_history, + normalizing_collections=normalizing_collections, + alg_name=alg_name, + alg_id=alg_id, + ) + overwrite_handled = True + + if mspass_object.live and not overwrite_handled: # We remove dead bodies from ensembles to simplify saving # data below. if isinstance(mspass_object, (TimeSeriesEnsemble, SeismogramEnsemble)): @@ -1635,7 +1713,7 @@ def save_data( alg_name, alg_id, ) - else: + elif not overwrite_handled: # may need to clean Metadata before calling this method # to assure there are not values that will cause MongoDB # to throw an exception when the tombstone subdocument @@ -3086,6 +3164,7 @@ def update_data( normalizing_collections=["channel", "site", "source"], alg_id="0", alg_name="Database.update_data", + save_history=True, ): """ Updates both metadata and sample data corresponding to an input data @@ -3108,11 +3187,13 @@ def update_data( this could create a blaat problem with large data sets so we do not currently support that type of update. - A VERY IMPORTANT implicit feature of this method is that - if the magic key "gridfs_id" exists the sample data in the - input to this method (mspass_object.data) will overwrite any - the existing content of gridfs found at the matching id. This - is a somewhat hidden feature so beware. + A VERY IMPORTANT feature of this method is that, when the magic + key ``gridfs_id`` exists, replacement samples are first written as + a new GridFS object. The waveform document is then changed from the + expected old id to the new id with one conditional update. Only + after that reference is durable is the old GridFS object deleted. + Consequently, a put or reference-update failure leaves the old + waveform readable instead of destroying its only sample copy. :param mspass_object: the object you want to update. :type mspass_object: either :class:`mspasspy.ccore.seismic.TimeSeries` or :class:`mspasspy.ccore.seismic.Seismogram` @@ -3149,6 +3230,11 @@ def update_data( (e.g. if one added a "shot" collection to the schema the list would need to be changed to at least add "shot".) :type normalizing_collections: list of strings defining collection names. + :param save_history: When True (the default), save a nonempty object + history and update its waveform cross-reference. False suppresses + that write. This option is used by :meth:`save_data` when its + documented GridFS overwrite mode updates an existing waveform. + :type save_history: :class:`bool` :param alg_name: alg_name is the name the func we are gonna save while preserving the history. (defaults to 'Database.update_data' and should not normally need to be changed) :type alg_name: :class:`str` @@ -3207,7 +3293,7 @@ def update_data( self.database_schema.default_name("history_object") + "_id" ) history_object_id = None - if not mspass_object.is_empty(): + if save_history and not mspass_object.is_empty(): history_object_id = self._save_history(mspass_object, alg_name, alg_id) update_record[history_obj_id_name] = history_object_id @@ -6798,11 +6884,13 @@ def _save_sample_data_to_gridfs( :type mspass_object: one of TimeSeries, Seismogram, TimeSeriesEnsemble, or Seismogram Ensemble. - :param overwrite: When set True if there is an existing datum - with a matching id for the attribute "gridfs_id", the existing datum - will be deleted before the new data is saved. When False a new - set of documents will be created to hold the data in the gridfs - system. Default is False. + :param overwrite: Compatibility flag used by higher-level writers. + This low-level method always stages a new GridFS object and never + deletes the object named by an existing ``gridfs_id``: it cannot + know whether a waveform reference has been committed. Public + atomic overwrite paths perform that conditional reference update + and delete the old object afterward. Direct callers must provide + the same ordering if they intend to remove old data. :param gridfs_id: optional id to assign to the newly written GridFS object. Used by update_data so a failed staged write can be rolled @@ -6816,10 +6904,6 @@ def _save_sample_data_to_gridfs( gfsh = gridfs.GridFS(self) if isinstance(mspass_object, (TimeSeries, Seismogram)): - if overwrite and mspass_object.is_defined("gridfs_id"): - old_gridfs_id = mspass_object["gridfs_id"] - if gfsh.exists(old_gridfs_id): - gfsh.delete(old_gridfs_id) # verdion 2 had a transpose here that seemed unncessary ub = bytes(np.array(mspass_object.data)) if gridfs_id is None: diff --git a/python/tests/db/test_database.py b/python/tests/db/test_database.py index 1571db282d..d59aee6036 100644 --- a/python/tests/db/test_database.py +++ b/python/tests/db/test_database.py @@ -430,9 +430,14 @@ def test_save_and_read_gridfs(self): gfsh = gridfs.GridFS(self.db) assert gfsh.exists(gridfs_id) - # test overwrite mode + # The low-level helper can only stage replacement data. It cannot + # delete the old object before a caller durably switches the waveform + # reference. self.db._save_sample_data_to_gridfs(tmp_ts, True) - assert not gfsh.exists(gridfs_id) + new_gridfs_id = tmp_ts["gridfs_id"] + assert new_gridfs_id != gridfs_id + assert gfsh.exists(gridfs_id) + assert gfsh.exists(new_gridfs_id) def mock_urlopen(*args): response = Mock() diff --git a/python/tests/db/test_gridfs_overwrite_contract.py b/python/tests/db/test_gridfs_overwrite_contract.py index 2b4e2455a6..90dbeb10c3 100644 --- a/python/tests/db/test_gridfs_overwrite_contract.py +++ b/python/tests/db/test_gridfs_overwrite_contract.py @@ -10,8 +10,10 @@ from mspasspy.ccore.seismic import ( DoubleVector, Seismogram, + SeismogramEnsemble, TimeReferenceType, TimeSeries, + TimeSeriesEnsemble, ) from mspasspy.ccore.utility import AtomicType, ErrorSeverity, MsPASSError, dmatrix from mspasspy.db.client import DBClient @@ -87,20 +89,37 @@ def save_original(database, factory): ] +def replace_samples(database, datum, entrypoint): + if entrypoint == "update_data": + return database.update_data(datum, mode="promiscuous") + return database.save_data( + datum, + mode="promiscuous", + storage_mode="gridfs", + overwrite=True, + save_history=False, + return_data=True, + ) + + @pytest.mark.parametrize("factory,collection", CASES) +@pytest.mark.parametrize("entrypoint", ["update_data", "save_data"]) def test_successful_overwrite_commits_new_reference_before_removing_old_blob( - database, factory, collection + database, factory, collection, entrypoint ): datum = save_original(database, factory) old_gridfs_id = datum["gridfs_id"] replacement = factory([5.0, 6.0, 7.0, 8.0]) datum.data = replacement.data - result = database.update_data(datum, mode="promiscuous") + result = replace_samples(database, datum, entrypoint) document = database[collection].find_one({"_id": datum["_id"]}) new_gridfs_id = document["gridfs_id"] assert result is datum + assert database[collection].count_documents({}) == 1 + if entrypoint == "save_data": + assert database["history_object"].count_documents({}) == 0 assert new_gridfs_id == datum["gridfs_id"] assert new_gridfs_id != old_gridfs_id storage = gridfs.GridFS(database) @@ -112,8 +131,9 @@ def test_successful_overwrite_commits_new_reference_before_removing_old_blob( @pytest.mark.parametrize("factory,collection", CASES) +@pytest.mark.parametrize("entrypoint", ["update_data", "save_data"]) def test_new_put_failure_preserves_old_reference_and_blob( - database, factory, collection + database, factory, collection, entrypoint ): datum = save_original(database, factory) old_gridfs_id = datum["gridfs_id"] @@ -129,7 +149,7 @@ def put_then_fail(self, data, *args, **kwargs): with patch.object(gridfs.GridFS, "put", put_then_fail): with pytest.raises(RuntimeError) as error: - database.update_data(datum, mode="promiscuous") + replace_samples(database, datum, entrypoint) assert error.value is failure assert datum["gridfs_id"] == old_gridfs_id @@ -143,8 +163,9 @@ def put_then_fail(self, data, *args, **kwargs): @pytest.mark.parametrize("factory,collection", CASES) +@pytest.mark.parametrize("entrypoint", ["update_data", "save_data"]) def test_reference_update_failure_removes_new_blob_and_rethrows_original_error( - database, factory, collection + database, factory, collection, entrypoint ): datum = save_original(database, factory) old_gridfs_id = datum["gridfs_id"] @@ -162,7 +183,7 @@ def fail_reference_update(self, query, update, *args, **kwargs): with patch.object(Collection, "update_one", fail_reference_update): with pytest.raises(RuntimeError) as error: - database.update_data(datum, mode="promiscuous") + replace_samples(database, datum, entrypoint) assert error.value is failure assert datum["gridfs_id"] == old_gridfs_id @@ -176,8 +197,9 @@ def fail_reference_update(self, query, update, *args, **kwargs): @pytest.mark.parametrize("factory,collection", CASES) +@pytest.mark.parametrize("entrypoint", ["update_data", "save_data"]) def test_reference_compare_miss_removes_new_blob_without_overwriting_current_reference( - database, factory, collection + database, factory, collection, entrypoint ): datum = save_original(database, factory) waveform_id = datum["_id"] @@ -196,7 +218,7 @@ def miss_reference_update(self, query, update, *args, **kwargs): with patch.object(Collection, "update_one", miss_reference_update): with pytest.raises(MsPASSError) as error: - database.update_data(datum, mode="promiscuous") + replace_samples(database, datum, entrypoint) assert error.value.severity == ErrorSeverity.Invalid assert final_queries == [{"_id": waveform_id, "gridfs_id": old_gridfs_id}] @@ -208,8 +230,9 @@ def miss_reference_update(self, query, update, *args, **kwargs): @pytest.mark.parametrize("factory,collection", CASES) +@pytest.mark.parametrize("entrypoint", ["update_data", "save_data"]) def test_old_delete_failure_keeps_committed_new_data_and_logs_one_complaint( - database, factory, collection + database, factory, collection, entrypoint ): datum = save_original(database, factory) old_gridfs_id = datum["gridfs_id"] @@ -225,7 +248,7 @@ def fail_old_delete(self, gridfs_id, *args, **kwargs): return original_delete(self, gridfs_id, *args, **kwargs) with patch.object(gridfs.GridFS, "delete", fail_old_delete): - result = database.update_data(datum, mode="promiscuous") + result = replace_samples(database, datum, entrypoint) document = database[collection].find_one({"_id": datum["_id"]}) new_gridfs_id = document["gridfs_id"] @@ -244,3 +267,53 @@ def fail_old_delete(self, gridfs_id, *args, **kwargs): reread = database.read_data(document, collection=collection) assert reread.live np.testing.assert_allclose(samples(reread), samples(replacement)) + + +@pytest.mark.parametrize( + "factory,ensemble_type,collection", + [ + (make_timeseries, TimeSeriesEnsemble, "wf_TimeSeries"), + (make_seismogram, SeismogramEnsemble, "wf_Seismogram"), + ], +) +def test_save_data_overwrite_updates_existing_ensemble_member_references( + database, factory, ensemble_type, collection +): + ensemble = ensemble_type() + ensemble.member.append(factory([1.0, 2.0, 3.0, 4.0])) + ensemble.member.append(factory([11.0, 12.0, 13.0, 14.0])) + ensemble.set_live() + ensemble = database.save_data( + ensemble, + storage_mode="gridfs", + save_history=False, + return_data=True, + ) + waveform_ids = [d["_id"] for d in ensemble.member] + old_gridfs_ids = [d["gridfs_id"] for d in ensemble.member] + replacements = [ + factory([5.0, 6.0, 7.0, 8.0]), + factory([15.0, 16.0, 17.0, 18.0]), + ] + for datum, replacement in zip(ensemble.member, replacements): + datum.data = replacement.data + + result = database.save_data( + ensemble, + mode="promiscuous", + storage_mode="gridfs", + overwrite=True, + save_history=False, + return_data=True, + ) + + ensemble = result + assert database[collection].count_documents({}) == 2 + storage = gridfs.GridFS(database) + for index, datum in enumerate(ensemble.member): + assert datum["_id"] == waveform_ids[index] + assert datum["gridfs_id"] != old_gridfs_ids[index] + assert not storage.exists(old_gridfs_ids[index]) + assert storage.exists(datum["gridfs_id"]) + reread = database.read_data(waveform_ids[index], collection=collection) + np.testing.assert_allclose(samples(reread), samples(replacements[index])) From d1335a1b164f9bc58a90fb4460fdbbb3437ee3dd Mon Sep 17 00:00:00 2001 From: wangyinz Date: Sat, 22 Aug 2026 15:34:15 -0500 Subject: [PATCH 3/5] Reject unsupported distributed GridFS overwrite --- python/mspasspy/io/distributed.py | 30 ++++++++----------- .../db/test_gridfs_overwrite_contract.py | 11 +++++++ 2 files changed, 23 insertions(+), 18 deletions(-) diff --git a/python/mspasspy/io/distributed.py b/python/mspasspy/io/distributed.py index 4cda6c3b44..e2222064f2 100644 --- a/python/mspasspy/io/distributed.py +++ b/python/mspasspy/io/distributed.py @@ -1431,16 +1431,12 @@ class method `save_data`. See the docstring for details but the more details on data export. Used only for "file" storage mode. :type file_format: :class:`str` - :param overwrite: If true gridfs data linked to the original - waveform will be replaced by the sample data from this save. - Default is false, and should be the normal use. This option - should never be used after a reduce operator as the parents - are not tracked and the space advantage is likely minimal for - the confusion it would cause. This is most useful for light, stable - preprocessing with a set of map operators to regularize a data - set before more extensive processing. It can only be used when - storage_mode is set to gridfs. - :type overwrite: boolean + :param overwrite: Distributed overwrite is not supported because this + writer bulk-inserts waveform documents and cannot atomically replace an + existing GridFS reference. Passing ``True`` raises ``ValueError`` before + any computation or persistent write. Use :meth:`Database.save_data` or + :meth:`Database.update_data` for failure-safe serial GridFS replacement. + :type overwrite: boolean :param collectiion: name of wf collection where the documents derived from the data are to be saved. Standard values are @@ -1550,6 +1546,12 @@ class method `save_data`. See the docstring for details but the "Must be one one of the following: promiscuous, cautious, or pedantic" ) raise ValueError(message) + if overwrite: + raise ValueError( + "write_distributed_data does not support overwrite=True; " + "use Database.save_data or Database.update_data for " + "failure-safe GridFS replacement" + ) if scheduler: if scheduler not in ["dask", "spark"]: message = "write_distributed_data: Illegal value of scheduler={}\n".format( @@ -1579,14 +1581,6 @@ class method `save_data`. See the docstring for details but the message += "Currently must be either wf_TimeSeries, wf_Seismogram, or default that implies wf_TimeSeries" raise ValueError(message) - if overwrite: - if storage_mode != "gridfs": - message = "write_distributed_data: overwrite mode is set True with storage_mode={}\n".format( - storage_mode - ) - message += "overwrite is only allowed with gridfs storage_mode" - raise ValueError(message) - stedronsky = Undertaker(db) if scheduler == "spark": if data_are_atomic: diff --git a/python/tests/db/test_gridfs_overwrite_contract.py b/python/tests/db/test_gridfs_overwrite_contract.py index 90dbeb10c3..0dad322546 100644 --- a/python/tests/db/test_gridfs_overwrite_contract.py +++ b/python/tests/db/test_gridfs_overwrite_contract.py @@ -19,6 +19,7 @@ from mspasspy.db.client import DBClient from mspasspy.db.collection import Collection from mspasspy.db.database import Database +from mspasspy.io.distributed import write_distributed_data @pytest.fixture @@ -317,3 +318,13 @@ def test_save_data_overwrite_updates_existing_ensemble_member_references( assert storage.exists(datum["gridfs_id"]) reread = database.read_data(waveform_ids[index], collection=collection) np.testing.assert_allclose(samples(reread), samples(replacements[index])) + + +@pytest.mark.parametrize("storage_mode", ["gridfs", "file"]) +def test_write_distributed_data_rejects_overwrite_before_execution(storage_mode): + database = object.__new__(Database) + + with pytest.raises(ValueError, match="does not support overwrite=True"): + write_distributed_data( + None, database, storage_mode=storage_mode, overwrite=True + ) From 1e4e68099698e2231914430d76e117cd8b07b242 Mon Sep 17 00:00:00 2001 From: wangyinz Date: Sat, 22 Aug 2026 15:57:29 -0500 Subject: [PATCH 4/5] Limit overwrite rollback to ordinary exceptions --- python/mspasspy/db/database.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/mspasspy/db/database.py b/python/mspasspy/db/database.py index 86aec5da45..5cb6cfd34b 100644 --- a/python/mspasspy/db/database.py +++ b/python/mspasspy/db/database.py @@ -3374,7 +3374,7 @@ def update_data( "GridFS reference", ErrorSeverity.Invalid, ) - except BaseException as original_error: + except Exception as original_error: if old_gridfs_id is not None: try: gfsh = gridfs.GridFS(self) From 019e2256e0fcdbc552e37bc42804fd1709fcb4f6 Mon Sep 17 00:00:00 2001 From: wangyinz Date: Sat, 22 Aug 2026 22:19:17 -0500 Subject: [PATCH 5/5] Update distributed overwrite error contract test --- python/tests/io/test_distributed.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/python/tests/io/test_distributed.py b/python/tests/io/test_distributed.py index 06c8425556..5bb62e3af9 100644 --- a/python/tests/io/test_distributed.py +++ b/python/tests/io/test_distributed.py @@ -1792,9 +1792,7 @@ def test_write_error_handlers(atomic_time_series_generator): with pytest.raises(ValueError, match="Illegal value of collection"): mybag = write_distributed_data(mybag, db, collection="illegal_collection") - with pytest.raises( - ValueError, match="overwrite mode is set True with storage_mode=" - ): + with pytest.raises(ValueError, match="does not support overwrite=True"): mybag = write_distributed_data( mybag, db, collection="wf_TimeSeries", overwrite=True, storage_mode="file" )