From e41b33269f1a467d839b61ec52a2652f6c2200bb Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Sat, 7 Feb 2026 02:01:58 +0100 Subject: [PATCH] Refactor --- changedetectionio/__init__.py | 10 +- changedetectionio/api/Notifications.py | 4 +- changedetectionio/api/Tags.py | 8 +- changedetectionio/api/Watch.py | 5 + .../blueprint/backups/__init__.py | 3 +- .../blueprint/price_data_follower/__init__.py | 2 + .../blueprint/settings/__init__.py | 13 +- changedetectionio/blueprint/tags/__init__.py | 7 +- changedetectionio/blueprint/ui/__init__.py | 12 +- changedetectionio/blueprint/ui/edit.py | 11 +- .../blueprint/watchlist/__init__.py | 2 +- changedetectionio/model/Watch.py | 33 ++ changedetectionio/store/__init__.py | 76 +-- changedetectionio/store/base.py | 8 +- .../store/file_saving_datastore.py | 548 +----------------- changedetectionio/store/updates.py | 12 +- changedetectionio/tests/conftest.py | 5 - changedetectionio/tests/util.py | 5 - 18 files changed, 146 insertions(+), 618 deletions(-) diff --git a/changedetectionio/__init__.py b/changedetectionio/__init__.py index ea45cd141..e19449cba 100644 --- a/changedetectionio/__init__.py +++ b/changedetectionio/__init__.py @@ -124,13 +124,9 @@ def sigshutdown_handler(_signo, _stack_frame): except Exception as e: logger.error(f"Error shutting down Socket.IO server: {str(e)}") - # Save data quickly - force immediate save using abstract method - try: - datastore.force_save_all() - logger.success('Fast sync to storage complete.') - except Exception as e: - logger.error(f"Error syncing to storage: {str(e)}") - + # With immediate persistence, all data is already saved + logger.success('All data already persisted (immediate commits enabled).') + sys.exit() def print_help(): diff --git a/changedetectionio/api/Notifications.py b/changedetectionio/api/Notifications.py index e25c6e9dd..7d2f8db98 100644 --- a/changedetectionio/api/Notifications.py +++ b/changedetectionio/api/Notifications.py @@ -67,7 +67,7 @@ class Notifications(Resource): clean_urls = [url.strip() for url in notification_urls if isinstance(url, str)] self.datastore.data['settings']['application']['notification_urls'] = clean_urls - self.datastore.needs_write = True + self.datastore.commit() return {'notification_urls': clean_urls}, 200 @@ -95,7 +95,7 @@ class Notifications(Resource): abort(400, message="No matching notification URLs found.") self.datastore.data['settings']['application']['notification_urls'] = notification_urls - self.datastore.needs_write = True + self.datastore.commit() return 'OK', 204 diff --git a/changedetectionio/api/Tags.py b/changedetectionio/api/Tags.py index e70d565af..ed4ee2661 100644 --- a/changedetectionio/api/Tags.py +++ b/changedetectionio/api/Tags.py @@ -63,9 +63,11 @@ class Tag(Resource): if request.args.get('muted', '') == 'muted': self.datastore.data['settings']['application']['tags'][uuid]['notification_muted'] = True + self.datastore.commit() return "OK", 200 elif request.args.get('muted', '') == 'unmuted': self.datastore.data['settings']['application']['tags'][uuid]['notification_muted'] = False + self.datastore.commit() return "OK", 200 return tag @@ -79,11 +81,13 @@ class Tag(Resource): # Delete the tag, and any tag reference del self.datastore.data['settings']['application']['tags'][uuid] - + self.datastore.commit() + # Remove tag from all watches for watch_uuid, watch in self.datastore.data['watching'].items(): if watch.get('tags') and uuid in watch['tags']: watch['tags'].remove(uuid) + watch.commit() return 'OK', 204 @@ -107,7 +111,7 @@ class Tag(Resource): return str(e), 400 tag.update(request.json) - self.datastore.needs_write_urgent = True + self.datastore.commit() return "OK", 200 diff --git a/changedetectionio/api/Watch.py b/changedetectionio/api/Watch.py index ac7ebcf41..e1433b7f7 100644 --- a/changedetectionio/api/Watch.py +++ b/changedetectionio/api/Watch.py @@ -84,15 +84,19 @@ class Watch(Resource): return "OK", 200 if request.args.get('paused', '') == 'paused': watch_obj.pause() + watch_obj.commit() return "OK", 200 elif request.args.get('paused', '') == 'unpaused': watch_obj.unpause() + watch_obj.commit() return "OK", 200 if request.args.get('muted', '') == 'muted': watch_obj.mute() + watch_obj.commit() return "OK", 200 elif request.args.get('muted', '') == 'unmuted': watch_obj.unmute() + watch_obj.commit() return "OK", 200 # Return without history, get that via another API call @@ -173,6 +177,7 @@ class Watch(Resource): # Update watch with regular (non-processor-config) fields watch.update(json_data) + watch.commit() # Save processor config to JSON file processors.save_processor_config(self.datastore, uuid, processor_config_data) diff --git a/changedetectionio/blueprint/backups/__init__.py b/changedetectionio/blueprint/backups/__init__.py index 646b80376..7e27562f4 100644 --- a/changedetectionio/blueprint/backups/__init__.py +++ b/changedetectionio/blueprint/backups/__init__.py @@ -102,8 +102,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): flash(gettext("Maximum number of backups reached, please remove some"), "error") return redirect(url_for('backups.index')) - # Be sure we're written fresh - force immediate save using abstract method - datastore.force_save_all() + # With immediate persistence, all data is already saved zip_thread = threading.Thread( target=create_backup, args=(datastore.datastore_path, datastore.data.get("watching")), diff --git a/changedetectionio/blueprint/price_data_follower/__init__.py b/changedetectionio/blueprint/price_data_follower/__init__.py index ecf60eea3..637f9856b 100644 --- a/changedetectionio/blueprint/price_data_follower/__init__.py +++ b/changedetectionio/blueprint/price_data_follower/__init__.py @@ -20,6 +20,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q: PriorityQueue datastore.data['watching'][uuid]['track_ldjson_price_data'] = PRICE_DATA_TRACK_ACCEPT datastore.data['watching'][uuid]['processor'] = 'restock_diff' datastore.data['watching'][uuid].clear_watch() + datastore.data['watching'][uuid].commit() worker_pool.queue_item_async_safe(update_q, queuedWatchMetaData.PrioritizedItem(priority=1, item={'uuid': uuid})) return redirect(url_for("watchlist.index")) @@ -27,6 +28,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q: PriorityQueue @price_data_follower_blueprint.route("//reject", methods=['GET']) def reject(uuid): datastore.data['watching'][uuid]['track_ldjson_price_data'] = PRICE_DATA_TRACK_REJECT + datastore.data['watching'][uuid].commit() return redirect(url_for("watchlist.index")) diff --git a/changedetectionio/blueprint/settings/__init__.py b/changedetectionio/blueprint/settings/__init__.py index 4c35ddb52..a68d0646f 100644 --- a/changedetectionio/blueprint/settings/__init__.py +++ b/changedetectionio/blueprint/settings/__init__.py @@ -74,12 +74,13 @@ def construct_blueprint(datastore: ChangeDetectionStore): del (app_update['password']) datastore.data['settings']['application'].update(app_update) - + # Handle dynamic worker count adjustment old_worker_count = datastore.data['settings']['requests'].get('workers', 1) new_worker_count = form.data['requests'].get('workers', 1) datastore.data['settings']['requests'].update(form.data['requests']) + datastore.commit() # Adjust worker count if it changed if new_worker_count != old_worker_count: @@ -109,13 +110,11 @@ def construct_blueprint(datastore: ChangeDetectionStore): if not os.getenv("SALTED_PASS", False) and len(form.application.form.password.encrypted_password): datastore.data['settings']['application']['password'] = form.application.form.password.encrypted_password - datastore.needs_write_urgent = True + datastore.commit() flash(gettext("Password protection enabled."), 'notice') flask_login.logout_user() return redirect(url_for('watchlist.index')) - datastore.needs_write_urgent = True - # Also save plugin settings from the same form submission plugin_tabs_list = get_plugin_settings_tabs() for tab in plugin_tabs_list: @@ -181,7 +180,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): def settings_reset_api_key(): secret = secrets.token_hex(16) datastore.data['settings']['application']['api_access_token'] = secret - datastore.needs_write_urgent = True + datastore.commit() flash(gettext("API Key was regenerated.")) return redirect(url_for('settings.settings_page')+'#api') @@ -198,7 +197,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): def toggle_all_paused(): current_state = datastore.data['settings']['application'].get('all_paused', False) datastore.data['settings']['application']['all_paused'] = not current_state - datastore.needs_write_urgent = True + datastore.commit() if datastore.data['settings']['application']['all_paused']: flash(gettext("Automatic scheduling paused - checks will not be queued."), 'notice') @@ -212,7 +211,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): def toggle_all_muted(): current_state = datastore.data['settings']['application'].get('all_muted', False) datastore.data['settings']['application']['all_muted'] = not current_state - datastore.needs_write_urgent = True + datastore.commit() if datastore.data['settings']['application']['all_muted']: flash(gettext("All notifications muted."), 'notice') diff --git a/changedetectionio/blueprint/tags/__init__.py b/changedetectionio/blueprint/tags/__init__.py index 761e93fe1..b16a9d411 100644 --- a/changedetectionio/blueprint/tags/__init__.py +++ b/changedetectionio/blueprint/tags/__init__.py @@ -59,6 +59,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): def mute(uuid): if datastore.data['settings']['application']['tags'].get(uuid): datastore.data['settings']['application']['tags'][uuid]['notification_muted'] = not datastore.data['settings']['application']['tags'][uuid]['notification_muted'] + datastore.commit() return redirect(url_for('tags.tags_overview_page')) @tags_blueprint.route("/delete/", methods=['GET']) @@ -76,6 +77,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): for watch_uuid, watch in datastore.data['watching'].items(): if watch.get('tags') and tag_uuid in watch['tags']: watch['tags'].remove(tag_uuid) + watch.commit() removed_count += 1 logger.info(f"Background: Tag {tag_uuid} removed from {removed_count} watches") except Exception as e: @@ -98,6 +100,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): for watch_uuid, watch in datastore.data['watching'].items(): if watch.get('tags') and tag_uuid in watch['tags']: watch['tags'].remove(tag_uuid) + watch.commit() unlinked_count += 1 logger.info(f"Background: Tag {tag_uuid} unlinked from {unlinked_count} watches") except Exception as e: @@ -114,6 +117,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): def delete_all(): # Clear all tags from settings immediately datastore.data['settings']['application']['tags'] = {} + datastore.commit() # Clear tags from all watches in background thread to avoid blocking def clear_all_tags_background(): @@ -122,6 +126,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): try: for watch_uuid, watch in datastore.data['watching'].items(): watch['tags'] = [] + watch.commit() cleared_count += 1 logger.info(f"Background: Cleared tags from {cleared_count} watches") except Exception as e: @@ -216,7 +221,7 @@ def construct_blueprint(datastore: ChangeDetectionStore): datastore.data['settings']['application']['tags'][uuid].update(form.data) datastore.data['settings']['application']['tags'][uuid]['processor'] = 'restock_diff' - datastore.needs_write_urgent = True + datastore.commit() flash(gettext("Updated")) return redirect(url_for('tags.tags_overview_page')) diff --git a/changedetectionio/blueprint/ui/__init__.py b/changedetectionio/blueprint/ui/__init__.py index 0606de2a7..75285d462 100644 --- a/changedetectionio/blueprint/ui/__init__.py +++ b/changedetectionio/blueprint/ui/__init__.py @@ -24,7 +24,7 @@ def _handle_operations(op, uuids, datastore, worker_pool, update_q, queuedWatchM for uuid in uuids: if datastore.data['watching'].get(uuid): datastore.data['watching'][uuid]['paused'] = True - datastore.mark_watch_dirty(uuid) + datastore.data['watching'][uuid].commit() if emit_flash: flash(gettext("{} watches paused").format(len(uuids))) @@ -32,7 +32,7 @@ def _handle_operations(op, uuids, datastore, worker_pool, update_q, queuedWatchM for uuid in uuids: if datastore.data['watching'].get(uuid): datastore.data['watching'][uuid.strip()]['paused'] = False - datastore.mark_watch_dirty(uuid) + datastore.data['watching'][uuid].commit() if emit_flash: flash(gettext("{} watches unpaused").format(len(uuids))) @@ -47,7 +47,7 @@ def _handle_operations(op, uuids, datastore, worker_pool, update_q, queuedWatchM for uuid in uuids: if datastore.data['watching'].get(uuid): datastore.data['watching'][uuid]['notification_muted'] = True - datastore.mark_watch_dirty(uuid) + datastore.data['watching'][uuid].commit() if emit_flash: flash(gettext("{} watches muted").format(len(uuids))) @@ -55,7 +55,7 @@ def _handle_operations(op, uuids, datastore, worker_pool, update_q, queuedWatchM for uuid in uuids: if datastore.data['watching'].get(uuid): datastore.data['watching'][uuid]['notification_muted'] = False - datastore.mark_watch_dirty(uuid) + datastore.data['watching'][uuid].commit() if emit_flash: flash(gettext("{} watches un-muted").format(len(uuids))) @@ -71,7 +71,7 @@ def _handle_operations(op, uuids, datastore, worker_pool, update_q, queuedWatchM for uuid in uuids: if datastore.data['watching'].get(uuid): datastore.data['watching'][uuid]["last_error"] = False - datastore.mark_watch_dirty(uuid) + datastore.data['watching'][uuid].commit() if emit_flash: flash(gettext("{} watches errors cleared").format(len(uuids))) @@ -92,6 +92,7 @@ def _handle_operations(op, uuids, datastore, worker_pool, update_q, queuedWatchM datastore.data['watching'][uuid]['notification_body'] = None datastore.data['watching'][uuid]['notification_urls'] = [] datastore.data['watching'][uuid]['notification_format'] = USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH + datastore.data['watching'][uuid].commit() if emit_flash: flash(gettext("{} watches set to use default notification settings").format(len(uuids))) @@ -107,6 +108,7 @@ def _handle_operations(op, uuids, datastore, worker_pool, update_q, queuedWatchM datastore.data['watching'][uuid]['tags'] = [] datastore.data['watching'][uuid]['tags'].append(tag_uuid) + datastore.data['watching'][uuid].commit() if emit_flash: flash(gettext("{} watches were tagged").format(len(uuids))) diff --git a/changedetectionio/blueprint/ui/edit.py b/changedetectionio/blueprint/ui/edit.py index d13497925..ce25ecd9e 100644 --- a/changedetectionio/blueprint/ui/edit.py +++ b/changedetectionio/blueprint/ui/edit.py @@ -198,6 +198,10 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe # Recast it if need be to right data Watch handler watch_class = processors.get_custom_watch_obj_for_processor(form.data.get('processor')) datastore.data['watching'][uuid] = watch_class(datastore_path=datastore.datastore_path, __datastore=datastore.data, default=datastore.data['watching'][uuid]) + + # Save the watch immediately + datastore.data['watching'][uuid].commit() + flash(gettext("Updated watch - unpaused!") if request.args.get('unpause_on_save') else gettext("Updated watch.")) # Cleanup any browsersteps session for this watch @@ -207,10 +211,6 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe except Exception as e: logger.debug(f"Error cleaning up browsersteps session: {e}") - # Re #286 - We wait for syncing new data to disk in another thread every 60 seconds - # But in the case something is added we should save straight away - datastore.needs_write_urgent = True - # Do not queue on edit if its not within the time range # @todo maybe it should never queue anyway on edit... @@ -386,6 +386,9 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe s = re.sub(r'[0-9]+', r'\\d+', s) datastore.data["watching"][uuid]['ignore_text'].append('/' + s + '/') + # Save the updated ignore_text + datastore.data["watching"][uuid].commit() + return f"Click to preview" return edit_blueprint \ No newline at end of file diff --git a/changedetectionio/blueprint/watchlist/__init__.py b/changedetectionio/blueprint/watchlist/__init__.py index 8010225bc..1191582b0 100644 --- a/changedetectionio/blueprint/watchlist/__init__.py +++ b/changedetectionio/blueprint/watchlist/__init__.py @@ -39,7 +39,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe elif op == 'mute': datastore.data['watching'][uuid].toggle_mute() - datastore.needs_write = True + datastore.data['watching'][uuid].commit() return redirect(url_for('watchlist.index', tag = active_tag_uuid)) # Sort by last_changed and add the uuid which is usually the key.. diff --git a/changedetectionio/model/Watch.py b/changedetectionio/model/Watch.py index 4a3040fa2..7adb9d521 100644 --- a/changedetectionio/model/Watch.py +++ b/changedetectionio/model/Watch.py @@ -844,6 +844,39 @@ class model(watch_base): def toggle_mute(self): self['notification_muted'] ^= True + def commit(self): + """ + Save this watch immediately to disk using atomic write. + + Replaces the old dirty-tracking system with immediate persistence. + Uses atomic write pattern (temp file + rename) for crash safety. + + Fire-and-forget: Logs errors but does not raise exceptions. + Watch data remains in memory even if save fails, so next commit will retry. + """ + from loguru import logger + + if not self.__datastore: + logger.error(f"Cannot commit watch {self.get('uuid')} without datastore reference") + return + + if not self.watch_data_dir: + logger.error(f"Cannot commit watch {self.get('uuid')} without datastore_path") + return + + # Convert to dict for serialization, excluding processor config keys + # Processor configs are stored separately in processor-specific JSON files + watch_dict = {k: v for k, v in dict(self).items() if not k.startswith('processor_config_')} + + # Use existing atomic write helper + from changedetectionio.store.file_saving_datastore import save_watch_atomic + try: + save_watch_atomic(self.watch_data_dir, self.get('uuid'), watch_dict) + logger.debug(f"Committed watch {self.get('uuid')}") + except Exception as e: + logger.error(f"Failed to commit watch {self.get('uuid')}: {e}") + + def extra_notification_token_values(self): # Used for providing extra tokens # return {'widget': 555} diff --git a/changedetectionio/store/__init__.py b/changedetectionio/store/__init__.py index 32ad54281..fa70b05a0 100644 --- a/changedetectionio/store/__init__.py +++ b/changedetectionio/store/__init__.py @@ -56,9 +56,7 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): # Should only be active for docker # logging.basicConfig(filename='/dev/stdout', level=logging.INFO) self.datastore_path = datastore_path - self.needs_write = False self.start_time = time.time() - self.stop_thread = False self.save_version_copy_json_db(version_tag) self.reload_state(datastore_path=datastore_path, include_default_watches=include_default_watches, version_tag=version_tag) @@ -286,19 +284,19 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): self.__data['app_guid'] = "test-" + str(uuid_builder.uuid4()) else: self.__data['app_guid'] = str(uuid_builder.uuid4()) - self.mark_settings_dirty() + self.commit() # Ensure RSS access token exists if not self.__data['settings']['application'].get('rss_access_token'): secret = secrets.token_hex(16) self.__data['settings']['application']['rss_access_token'] = secret - self.mark_settings_dirty() + self.commit() # Ensure API access token exists if not self.__data['settings']['application'].get('api_access_token'): secret = secrets.token_hex(16) self.__data['settings']['application']['api_access_token'] = secret - self.mark_settings_dirty() + self.commit() # Handle password reset lockfile password_reset_lockfile = os.path.join(self.datastore_path, "removepassword.lock") @@ -306,9 +304,6 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): self.remove_password() unlink(password_reset_lockfile) - # Start the background save thread - self.start_save_thread() - def rehydrate_entity(self, uuid, entity, processor_override=None): """Set the dict back to the dict Watch object""" entity['uuid'] = uuid @@ -375,22 +370,15 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): Implementation of abstract method from FileSavingDataStore. Delegates to helper function and stores results in internal data structure. """ - watching, watch_hashes = load_all_watches( + watching = load_all_watches( self.datastore_path, - self.rehydrate_entity, - self._compute_hash + self.rehydrate_entity ) # Store loaded data self.__data['watching'] = watching - self._watch_hashes = watch_hashes - # Verify all watches have hashes - missing_hashes = [uuid for uuid in watching.keys() if uuid not in watch_hashes] - if missing_hashes: - logger.error(f"WARNING: {len(missing_hashes)} watches missing hashes after load: {missing_hashes[:5]}") - else: - logger.debug(f"All {len(watching)} watches have valid hashes") + logger.debug(f"Loaded {len(watching)} watches") def _delete_watch(self, uuid): """ @@ -414,7 +402,7 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): def set_last_viewed(self, uuid, timestamp): logger.debug(f"Setting watch UUID: {uuid} last viewed to {int(timestamp)}") self.data['watching'][uuid].update({'last_viewed': int(timestamp)}) - self.mark_watch_dirty(uuid) + self.data['watching'][uuid].commit() watch_check_update = signal('watch_check_update') if watch_check_update: @@ -422,7 +410,23 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): def remove_password(self): self.__data['settings']['application']['password'] = False - self.mark_settings_dirty() + self.commit() + + def commit(self): + """ + Save settings immediately to disk using atomic write. + + Replaces the old mark_settings_dirty() system with immediate persistence. + Uses atomic write pattern (temp file + rename) for crash safety. + + Fire-and-forget: Logs errors but does not raise exceptions. + Settings data remains in memory even if save fails, so next commit will retry. + """ + try: + self._save_settings() + logger.debug("Committed settings") + except Exception as e: + logger.error(f"Failed to commit settings: {e}") def update_watch(self, uuid, update_obj): @@ -441,7 +445,8 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): self.__data['watching'][uuid].update(update_obj) - self.mark_watch_dirty(uuid) + # Immediate save + self.__data['watching'][uuid].commit() @property def threshold_seconds(self): @@ -502,10 +507,6 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): except Exception as e: logger.error(f"Failed to delete watch {watch_uuid} from storage: {e}") - # Clean up tracking data - self._watch_hashes.pop(watch_uuid, None) - self._dirty_watches.discard(watch_uuid) - # Send delete signal watch_delete_signal = signal('watch_deleted') if watch_delete_signal: @@ -527,17 +528,11 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): # Remove from watching dict del self.data['watching'][uuid] - # Clean up tracking data - self._watch_hashes.pop(uuid, None) - self._dirty_watches.discard(uuid) - # Send delete signal watch_delete_signal = signal('watch_deleted') if watch_delete_signal: watch_delete_signal.send(watch_uuid=uuid) - self.needs_write_urgent = True - # Clone a watch by UUID def clone(self, uuid): url = self.data['watching'][uuid].get('url') @@ -562,7 +557,7 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): # Remove a watchs data but keep the entry (URL etc) def clear_watch_history(self, uuid): self.__data['watching'][uuid].clear_watch() - self.needs_write_urgent = True + self.__data['watching'][uuid].commit() def add_watch(self, url, tag='', extras=None, tag_uuids=None, save_immediately=True): @@ -675,16 +670,9 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): self.__data['watching'][new_uuid] = new_watch if save_immediately: - # Save immediately using polymorphic method - try: - self.save_watch(new_uuid, force=True) - logger.debug(f"Saved new watch {new_uuid}") - except Exception as e: - logger.error(f"Failed to save new watch {new_uuid}: {e}") - # Mark dirty for retry - self.mark_watch_dirty(new_uuid) - else: - self.mark_watch_dirty(new_uuid) + # Save immediately using commit + new_watch.commit() + logger.debug(f"Saved new watch {new_uuid}") logger.debug(f"Added '{url}'") @@ -889,7 +877,7 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): self.__data['settings']['application']['tags'][new_uuid] = new_tag - self.mark_settings_dirty() + self.commit() return new_uuid def get_all_tags_for_watch(self, uuid): @@ -1006,7 +994,7 @@ class ChangeDetectionStore(DatastoreUpdatesMixin, FileSavingDataStore): notification_urls.append(notification_url) self.__data['settings']['application']['notification_urls'] = notification_urls - self.mark_settings_dirty() + self.commit() return notification_url # Schema update methods moved to store/updates.py (DatastoreUpdatesMixin) diff --git a/changedetectionio/store/base.py b/changedetectionio/store/base.py index c0c9e47c1..5627e4d9e 100644 --- a/changedetectionio/store/base.py +++ b/changedetectionio/store/base.py @@ -88,13 +88,11 @@ class DataStore(ABC): This is the abstract method for forcing a complete save. Different backends implement this differently: - - File backend: Mark all watches/settings dirty, then save + - File backend: No-op (watches save immediately via commit()) - Redis backend: SAVE command or pipeline flush - SQL backend: COMMIT transaction - Used by: - - Backup creation (ensure everything is saved before backup) - - Shutdown (ensure all changes are persisted) - - Manual save operations + Note: With immediate persistence, this is mostly a no-op for file backend. + Used for backward compatibility. """ pass diff --git a/changedetectionio/store/file_saving_datastore.py b/changedetectionio/store/file_saving_datastore.py index f1363100e..aa17c6c2d 100644 --- a/changedetectionio/store/file_saving_datastore.py +++ b/changedetectionio/store/file_saving_datastore.py @@ -1,12 +1,11 @@ """ -File-based datastore with individual watch persistence and dirty tracking. +File-based datastore with individual watch persistence and immediate commits. This module provides the FileSavingDataStore abstract class that implements: - Individual watch.json file persistence -- Hash-based change detection (only save what changed) -- Periodic audit scan (catches unmarked changes) -- Background save thread with batched parallel saves +- Immediate commit-based persistence (watch.commit(), datastore.commit()) - Atomic file writes safe for NFS/NAS +- Backward compatibility stubs for legacy methods """ import glob @@ -34,18 +33,6 @@ except ImportError: # Set to True for mission-critical deployments requiring crash consistency FORCE_FSYNC_DATA_IS_CRITICAL = bool(strtobool(os.getenv('FORCE_FSYNC_DATA_IS_CRITICAL', 'False'))) -# Save interval configuration: How often the background thread saves dirty items -# Default 10 seconds - increase for less frequent saves, decrease for more frequent -DATASTORE_SCAN_DIRTY_SAVE_INTERVAL_SECONDS = int(os.getenv('DATASTORE_SCAN_DIRTY_SAVE_INTERVAL_SECONDS', '10')) - -# Rolling audit configuration: Scans a fraction of watches each cycle -# Default: Run audit every 10s, split into 5 shards -# Full audit completes every 50s (10s × 5 shards) -# With 56k watches: 56k / 5 = ~11k watches per cycle (~60ms vs 316ms for all) -# Handles dynamic watch count - recalculates shard boundaries each cycle -DATASTORE_AUDIT_INTERVAL_SECONDS = int(os.getenv('DATASTORE_AUDIT_INTERVAL_SECONDS', '10')) -DATASTORE_AUDIT_SHARDS = int(os.getenv('DATASTORE_AUDIT_SHARDS', '5')) - # ============================================================================ # Helper Functions for Atomic File Operations @@ -242,11 +229,6 @@ def load_watch_from_file(watch_json, uuid, rehydrate_entity_func): with open(watch_json, 'r', encoding='utf-8') as f: watch_data = json.load(f) - if watch_data.get('time_schedule_limit'): - del watch_data['time_schedule_limit'] - if watch_data.get('time_between_check'): - del watch_data['time_between_check'] - # Return both the raw data and the rehydrated watch # Raw data is needed to compute hash before rehydration changes anything watch_obj = rehydrate_entity_func(uuid, watch_data) @@ -278,7 +260,7 @@ def load_watch_from_file(watch_json, uuid, rehydrate_entity_func): return None, None -def load_all_watches(datastore_path, rehydrate_entity_func, compute_hash_func): +def load_all_watches(datastore_path, rehydrate_entity_func): """ Load all watches from individual watch.json files. @@ -289,21 +271,17 @@ def load_all_watches(datastore_path, rehydrate_entity_func, compute_hash_func): Args: datastore_path: Path to the datastore directory rehydrate_entity_func: Function to convert dict to Watch object - compute_hash_func: Function to compute hash from raw watch dict Returns: - Tuple of (watching_dict, hashes_dict) - - watching_dict: uuid -> Watch object - - hashes_dict: uuid -> hash string (computed from raw data) + Dictionary of uuid -> Watch object """ start_time = time.time() logger.info("Loading watches from individual watch.json files...") watching = {} - watch_hashes = {} if not os.path.exists(datastore_path): - return watching, watch_hashes + return watching # Find all watch.json files using glob (faster than manual directory traversal) glob_start = time.time() @@ -322,9 +300,6 @@ def load_all_watches(datastore_path, rehydrate_entity_func, compute_hash_func): watch, raw_data = load_watch_from_file(watch_json, uuid_dir, rehydrate_entity_func) if watch and raw_data: watching[uuid_dir] = watch - # Compute hash from rehydrated Watch object (as dict) to match how we compute on save - # This ensures hash matches what audit will compute from dict(watch) - watch_hashes[uuid_dir] = compute_hash_func(dict(watch)) loaded += 1 if loaded % 100 == 0: @@ -344,7 +319,7 @@ def load_all_watches(datastore_path, rehydrate_entity_func, compute_hash_func): else: logger.info(f"Loaded {loaded} watches from disk in {elapsed:.2f}s ({loaded/elapsed:.0f} watches/sec)") - return watching, watch_hashes + return watching # ============================================================================ @@ -353,151 +328,29 @@ def load_all_watches(datastore_path, rehydrate_entity_func, compute_hash_func): class FileSavingDataStore(DataStore): """ - Abstract datastore that provides file persistence with change tracking. + Abstract datastore that provides file persistence with immediate commits. Features: - Individual watch.json files (one per watch) - - Dirty tracking: Only saves items that have changed - - Hash-based change detection: Prevents unnecessary writes - - Background save thread: Non-blocking persistence - - Two-tier urgency: Standard (60s) and urgent (immediate) saves + - Immediate persistence via watch.commit() and datastore.commit() + - Atomic file writes for crash safety + - Backward compatibility stubs for legacy methods Subclasses must implement: - rehydrate_entity(): Convert dict to Watch object - Access to internal __data structure for watch management """ - needs_write = False - needs_write_urgent = False - stop_thread = False - - # Change tracking - _dirty_watches = set() # Watch UUIDs that need saving - _dirty_settings = False # Settings changed - _watch_hashes = {} # UUID -> SHA256 hash for change detection - - # Health monitoring - _last_save_time = 0 # Timestamp of last successful save - _last_audit_time = 0 # Timestamp of last audit scan - _save_cycle_count = 0 # Number of save cycles completed - _total_saves = 0 # Total watches saved (lifetime) - _save_errors = 0 # Total save errors (lifetime) - _audit_count = 0 # Number of audit scans completed - _audit_found_changes = 0 # Total unmarked changes found by audits - _audit_shard_index = 0 # Current shard being audited (rolling audit) - def __init__(self): super().__init__() - self.save_data_thread = None - self._last_save_time = time.time() - self._last_audit_time = time.time() - def mark_watch_dirty(self, uuid): - """ - Mark a watch as needing save. - - Args: - uuid: Watch UUID - """ - with self.lock: - self._dirty_watches.add(uuid) - dirty_count = len(self._dirty_watches) - - # Backpressure detection - warn if dirty set grows too large - if dirty_count > 1000: - logger.critical( - f"BACKPRESSURE WARNING: {dirty_count} watches pending save! " - f"Save thread may not be keeping up with write rate. " - f"This could indicate disk I/O bottleneck or save thread failure." - ) - elif dirty_count > 500: - logger.warning( - f"Dirty watch count high: {dirty_count} watches pending save. " - f"Monitoring for potential backpressure." - ) - - self.needs_write = True + """Deprecated: Watches now save immediately via commit().""" + pass def mark_settings_dirty(self): - """Mark settings as needing save.""" - with self.lock: - self._dirty_settings = True - self.needs_write = True - - def _compute_hash(self, watch_dict): - """ - Compute SHA256 hash of watch for change detection. - - Args: - watch_dict: Dictionary representation of watch - - Returns: - Hex string of SHA256 hash - """ - # Use orjson for deterministic serialization if available - if HAS_ORJSON: - json_bytes = orjson.dumps(watch_dict, option=orjson.OPT_SORT_KEYS) - else: - json_str = json.dumps(watch_dict, sort_keys=True, ensure_ascii=False) - json_bytes = json_str.encode('utf-8') - - return hashlib.sha256(json_bytes).hexdigest() - - def save_watch(self, uuid, force=False, watch_dict=None, current_hash=None): - """ - Save a single watch if it has changed (polymorphic method). - - Args: - uuid: Watch UUID - force: If True, skip hash check and save anyway - watch_dict: Pre-computed watch dictionary (optimization) - current_hash: Pre-computed hash (optimization) - - Returns: - True if saved, False if skipped (unchanged) - """ - if not self._watch_exists(uuid): - logger.warning(f"Cannot save watch {uuid} - does not exist") - return False - - # Get watch dict if not provided - if watch_dict is None: - watch_dict = self._get_watch_dict(uuid) - - # Compute hash if not provided - if current_hash is None: - current_hash = self._compute_hash(watch_dict) - - # Skip save if unchanged (unless forced) - if not force and current_hash == self._watch_hashes.get(uuid): - return False - - try: - self._save_watch(uuid, watch_dict) - self._watch_hashes[uuid] = current_hash - logger.debug(f"Saved watch {uuid}") - return True - except Exception as e: - logger.error(f"Failed to save watch {uuid}: {e}") - raise - - def _save_watch(self, uuid, watch_dict): - """ - Save a single watch to storage (polymorphic). - - Backend-specific implementation. Subclasses override for different storage: - - File backend: Writes to {uuid}/watch.json - - Redis backend: SET watch:{uuid} - - SQL backend: UPDATE watches WHERE uuid=? - - Args: - uuid: Watch UUID - watch_dict: Dictionary representation of watch - """ - # Default file implementation - watch_dir = os.path.join(self.datastore_path, uuid) - save_watch_atomic(watch_dir, uuid, watch_dict) + """Deprecated: Settings now save immediately.""" + pass def _save_settings(self): """ @@ -510,6 +363,17 @@ class FileSavingDataStore(DataStore): """ raise NotImplementedError("Subclass must implement _save_settings") + def force_save_all(self): + """ + Deprecated: Watches now save immediately via commit(). + + Kept as no-op for backward compatibility. In the old system, + this would force all watches to save. With immediate persistence, + all changes are already saved. + """ + logger.info("force_save_all() called - no-op (immediate persistence enabled)") + pass + def _load_watches(self): """ Load all watches from storage (polymorphic). @@ -535,364 +399,4 @@ class FileSavingDataStore(DataStore): """ raise NotImplementedError("Subclass must implement _delete_watch") - def _save_dirty_items(self): - """ - Save dirty watches and settings. - This is the core optimization: instead of saving the entire datastore, - we only save watches that were marked dirty and settings if changed. - """ - start_time = time.time() - - # Capture dirty sets under lock - with self.lock: - dirty_watches = list(self._dirty_watches) - dirty_settings = self._dirty_settings - self._dirty_watches.clear() - self._dirty_settings = False - - if not dirty_watches and not dirty_settings: - return - - logger.trace(f"Saving {len(dirty_watches)} dirty watches, settings_dirty={dirty_settings}") - - # Save each dirty watch using the polymorphic save method - saved_count = 0 - error_count = 0 - skipped_unchanged = 0 - - # Process in batches of 50, using thread pool for parallel saves - BATCH_SIZE = 50 - MAX_WORKERS = 20 # Number of parallel save threads - - def save_single_watch(uuid): - """Helper function for thread pool execution.""" - try: - # Check if watch still exists (might have been deleted) - if not self._watch_exists(uuid): - # Watch was deleted, remove hash - self._watch_hashes.pop(uuid, None) - return {'status': 'deleted', 'uuid': uuid} - - # Pre-check hash to avoid unnecessary save_watch() calls - watch_dict = self._get_watch_dict(uuid) - current_hash = self._compute_hash(watch_dict) - - if current_hash == self._watch_hashes.get(uuid): - # Watch hasn't actually changed, skip - return {'status': 'unchanged', 'uuid': uuid} - - # Pass pre-computed values to avoid redundant serialization/hashing - if self.save_watch(uuid, force=True, watch_dict=watch_dict, current_hash=current_hash): - return {'status': 'saved', 'uuid': uuid} - else: - return {'status': 'skipped', 'uuid': uuid} - except Exception as e: - logger.error(f"Error saving watch {uuid}: {e}") - return {'status': 'error', 'uuid': uuid, 'error': e} - - # Process dirty watches in batches - for batch_start in range(0, len(dirty_watches), BATCH_SIZE): - batch = dirty_watches[batch_start:batch_start + BATCH_SIZE] - batch_num = (batch_start // BATCH_SIZE) + 1 - total_batches = (len(dirty_watches) + BATCH_SIZE - 1) // BATCH_SIZE - - if len(dirty_watches) > BATCH_SIZE: - logger.trace(f"Save batch {batch_num}/{total_batches} ({len(batch)} watches)") - - # Use thread pool to save watches in parallel - with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: - # Submit all save tasks - future_to_uuid = {executor.submit(save_single_watch, uuid): uuid for uuid in batch} - - # Collect results as they complete - for future in as_completed(future_to_uuid): - result = future.result() - status = result['status'] - - if status == 'saved': - saved_count += 1 - elif status == 'unchanged': - skipped_unchanged += 1 - elif status == 'error': - error_count += 1 - # Re-mark for retry - with self.lock: - self._dirty_watches.add(result['uuid']) - # 'deleted' and 'skipped' don't need special handling - - # Save settings if changed - if dirty_settings: - try: - self._save_settings() - logger.debug("Saved settings") - except Exception as e: - logger.error(f"Failed to save settings: {e}") - error_count += 1 - with self.lock: - self._dirty_settings = True - - # Update metrics - elapsed = time.time() - start_time - self._save_cycle_count += 1 - self._total_saves += saved_count - self._save_errors += error_count - self._last_save_time = time.time() - - # Log performance metrics - if saved_count > 0: - avg_time_per_watch = (elapsed / saved_count) * 1000 # milliseconds - skipped_msg = f", {skipped_unchanged} unchanged" if skipped_unchanged > 0 else "" - parallel_msg = f" [parallel: {MAX_WORKERS} workers]" if saved_count > 1 else "" - logger.info( - f"Successfully saved {saved_count} watches in {elapsed:.2f}s " - f"(avg {avg_time_per_watch:.1f}ms per watch{skipped_msg}){parallel_msg}. " - f"Total: {self._total_saves} saves, {self._save_errors} errors (lifetime)" - ) - elif skipped_unchanged > 0: - logger.debug(f"Save cycle: {skipped_unchanged} watches verified unchanged (hash match), nothing saved") - - if error_count > 0: - logger.error(f"Save cycle completed with {error_count} errors") - - self.needs_write = False - self.needs_write_urgent = False - - def _watch_exists(self, uuid): - """ - Check if watch exists. Subclass must implement. - - Args: - uuid: Watch UUID - - Returns: - bool - """ - raise NotImplementedError("Subclass must implement _watch_exists") - - def _get_watch_dict(self, uuid): - """ - Get watch as dictionary. Subclass must implement. - - Args: - uuid: Watch UUID - - Returns: - Dictionary representation of watch - """ - raise NotImplementedError("Subclass must implement _get_watch_dict") - - def _audit_all_watches(self): - """ - Rolling audit: Scans a fraction of watches to detect unmarked changes. - - Instead of scanning ALL watches at once, this scans 1/N shards per cycle. - The shard rotates each cycle, completing a full audit every N cycles. - - Handles dynamic watch count - recalculates shard boundaries each cycle, - so newly added watches will be audited in subsequent cycles. - - Benefits: - - Lower CPU per cycle (56k / 5 = ~11k watches vs all 56k) - - More frequent audits overall (every 50s vs every 10s) - - Spreads load evenly across time - """ - audit_start = time.time() - - # Get list of all watch UUIDs (read-only, no lock needed) - try: - all_uuids = list(self.data['watching'].keys()) - except (KeyError, AttributeError, RuntimeError): - # Data structure not ready or being modified - return - - if not all_uuids: - return - - total_watches = len(all_uuids) - - # Calculate this cycle's shard boundaries - # Example: 56,278 watches / 5 shards = 11,255 watches per shard - # Shard 0: [0:11255], Shard 1: [11255:22510], etc. - shard_size = (total_watches + DATASTORE_AUDIT_SHARDS - 1) // DATASTORE_AUDIT_SHARDS - start_idx = self._audit_shard_index * shard_size - end_idx = min(start_idx + shard_size, total_watches) - - # Handle wrap-around (shouldn't happen normally, but defensive) - if start_idx >= total_watches: - self._audit_shard_index = 0 - start_idx = 0 - end_idx = min(shard_size, total_watches) - - # Audit only this shard's watches - shard_uuids = all_uuids[start_idx:end_idx] - - changes_found = 0 - errors = 0 - - for uuid in shard_uuids: - try: - # Get current watch dict and compute hash - watch_dict = self._get_watch_dict(uuid) - current_hash = self._compute_hash(watch_dict) - stored_hash = self._watch_hashes.get(uuid) - - # If hash changed and not already marked dirty, mark it - if current_hash != stored_hash: - with self.lock: - if uuid not in self._dirty_watches: - self._dirty_watches.add(uuid) - changes_found += 1 - logger.warning( - f"Audit detected unmarked change in watch {uuid[:8]}... current {current_hash:8} stored hash {stored_hash[:8]}" - f"(hash changed but not marked dirty)" - ) - self.needs_write = True - except Exception as e: - errors += 1 - logger.trace(f"Audit error for watch {uuid[:8]}...: {e}") - - audit_elapsed = (time.time() - audit_start) * 1000 # milliseconds - - # Advance to next shard (wrap around after last shard) - self._audit_shard_index = (self._audit_shard_index + 1) % DATASTORE_AUDIT_SHARDS - - # Update metrics - self._audit_count += 1 - self._audit_found_changes += changes_found - self._last_audit_time = time.time() - - if changes_found > 0: - logger.warning( - f"Audit shard {self._audit_shard_index}/{DATASTORE_AUDIT_SHARDS} found {changes_found} " - f"unmarked changes in {len(shard_uuids)}/{total_watches} watches ({audit_elapsed:.1f}ms)" - ) - else: - logger.trace( - f"Audit shard {self._audit_shard_index}/{DATASTORE_AUDIT_SHARDS}: " - f"{len(shard_uuids)}/{total_watches} watches checked, 0 changes ({audit_elapsed:.1f}ms)" - ) - - def save_datastore(self): - """ - Background thread that periodically saves dirty items and audits watches. - - Runs two independent cycles: - 1. Save dirty items every DATASTORE_SCAN_DIRTY_SAVE_INTERVAL_SECONDS (default 10s) - 2. Rolling audit: every DATASTORE_AUDIT_INTERVAL_SECONDS (default 10s) - - Scans 1/DATASTORE_AUDIT_SHARDS watches per cycle (default 1/5) - - Full audit completes every 50s (10s × 5 shards) - - Automatically handles new/deleted watches - - Uses 0.5s sleep intervals for responsiveness to urgent saves. - """ - while True: - if self.stop_thread: - # Graceful shutdown: flush any remaining dirty items before stopping - if self.needs_write or self._dirty_watches or self._dirty_settings: - logger.warning("Datastore save thread stopping - flushing remaining dirty items...") - try: - self._save_dirty_items() - logger.info("Graceful shutdown complete - all data saved") - except Exception as e: - logger.critical(f"FAILED to save dirty items during shutdown: {e}") - else: - logger.info("Datastore save thread stopping - no dirty items") - return - - # Check if it's time to run audit scan (every N seconds) - if time.time() - self._last_audit_time >= DATASTORE_AUDIT_INTERVAL_SECONDS: - try: - self._audit_all_watches() - except Exception as e: - logger.error(f"Error in audit cycle: {e}") - - # Save dirty items if needed - if self.needs_write or self.needs_write_urgent: - try: - self._save_dirty_items() - except Exception as e: - logger.error(f"Error in save cycle: {e}") - - # Timer with early break for urgent saves - # Each iteration is 0.5 seconds, so iterations = DATASTORE_SCAN_DIRTY_SAVE_INTERVAL_SECONDS * 2 - for i in range(DATASTORE_SCAN_DIRTY_SAVE_INTERVAL_SECONDS * 2): - time.sleep(0.5) - if self.stop_thread or self.needs_write_urgent: - break - - def start_save_thread(self): - """Start the background save thread.""" - if not self.save_data_thread or not self.save_data_thread.is_alive(): - self.save_data_thread = Thread(target=self.save_datastore, daemon=True, name="DatastoreSaver") - self.save_data_thread.start() - logger.info("Datastore save thread started") - - def force_save_all(self): - """ - Force immediate synchronous save of all changes to storage. - - File backend implementation of the abstract force_save_all() method. - Marks all watches and settings as dirty, then saves immediately. - - Used by: - - Backup creation (ensure everything is saved before backup) - - Shutdown (ensure all changes are persisted) - - Manual save operations - """ - logger.info("Force saving all data to storage...") - - # Mark everything as dirty to ensure complete save - for uuid in self.data['watching'].keys(): - self.mark_watch_dirty(uuid) - self.mark_settings_dirty() - - # Save immediately (synchronous) - self._save_dirty_items() - - logger.success("All data saved to storage") - - def get_health_status(self): - """ - Get datastore health status for monitoring. - - Returns: - dict with health metrics and status - """ - now = time.time() - time_since_last_save = now - self._last_save_time - - with self.lock: - dirty_count = len(self._dirty_watches) - - is_thread_alive = self.save_data_thread and self.save_data_thread.is_alive() - - # Determine health status - if not is_thread_alive: - status = "CRITICAL" - message = "Save thread is DEAD" - elif time_since_last_save > 300: # 5 minutes - status = "WARNING" - message = f"No save activity for {time_since_last_save:.0f}s" - elif dirty_count > 1000: - status = "WARNING" - message = f"High backpressure: {dirty_count} watches pending" - elif self._save_errors > 0 and (self._save_errors / max(self._total_saves, 1)) > 0.01: - status = "WARNING" - message = f"High error rate: {self._save_errors} errors" - else: - status = "HEALTHY" - message = "Operating normally" - - return { - "status": status, - "message": message, - "thread_alive": is_thread_alive, - "dirty_watches": dirty_count, - "dirty_settings": self._dirty_settings, - "last_save_seconds_ago": int(time_since_last_save), - "save_cycles": self._save_cycle_count, - "total_saves": self._total_saves, - "total_errors": self._save_errors, - "error_rate_percent": round((self._save_errors / max(self._total_saves, 1)) * 100, 2) - } diff --git a/changedetectionio/store/updates.py b/changedetectionio/store/updates.py index 371253255..3634f0181 100644 --- a/changedetectionio/store/updates.py +++ b/changedetectionio/store/updates.py @@ -168,7 +168,7 @@ class DatastoreUpdatesMixin: latest_update = updates_available[-1] if updates_available else 0 logger.info(f"No schema version found and no watches exist - assuming fresh install, setting schema_version to {latest_update}") self.data['settings']['application']['schema_version'] = latest_update - self.mark_settings_dirty() + self.commit() return # No updates needed for fresh install else: # Has watches but no schema version - likely old datastore, run all updates @@ -201,14 +201,14 @@ class DatastoreUpdatesMixin: else: # Bump the version, important self.data['settings']['application']['schema_version'] = update_n - self.mark_settings_dirty() + self.commit() - # CRITICAL: Mark all watches as dirty so changes are persisted + # CRITICAL: Save all watches so changes are persisted # Most updates modify watches, and in the new individual watch.json structure, # we need to ensure those changes are saved - logger.info(f"Marking all {len(self.data['watching'])} watches as dirty after update_{update_n} (so that it saves them to disk)") + logger.info(f"Saving all {len(self.data['watching'])} watches after update_{update_n} (so that it saves them to disk)") for uuid in self.data['watching'].keys(): - self.mark_watch_dirty(uuid) + self.data['watching'][uuid].commit() # Save changes immediately after each update (more resilient than batching) logger.critical(f"Saving all changes after update_{update_n}") @@ -662,7 +662,7 @@ class DatastoreUpdatesMixin: updates_available = self.get_updates_available() latest_schema = updates_available[-1] if updates_available else 26 self.data['settings']['application']['schema_version'] = latest_schema - self.mark_settings_dirty() + self.commit() logger.info(f"Set schema_version to {latest_schema} (migration complete, all watches already saved)") logger.critical("=" * 80) diff --git a/changedetectionio/tests/conftest.py b/changedetectionio/tests/conftest.py index 5a71bcf5d..9ef4a41b2 100644 --- a/changedetectionio/tests/conftest.py +++ b/changedetectionio/tests/conftest.py @@ -308,10 +308,6 @@ def prepare_test_function(live_server, datastore_path): - # Prevent background thread from writing during cleanup/reload - datastore.needs_write = False - datastore.needs_write_urgent = False - # CRITICAL: Clean up any files from previous tests # This ensures a completely clean directory cleanup(datastore_path) @@ -344,7 +340,6 @@ def prepare_test_function(live_server, datastore_path): break datastore.data['watching'] = {} - datastore.needs_write = True except Exception as e: logger.warning(f"Error during datastore cleanup: {e}") diff --git a/changedetectionio/tests/util.py b/changedetectionio/tests/util.py index d79d3c74a..81dfdc8d6 100644 --- a/changedetectionio/tests/util.py +++ b/changedetectionio/tests/util.py @@ -161,11 +161,6 @@ def extract_UUID_from_client(client): def delete_all_watches(client=None): - # Change tracking - client.application.config.get('DATASTORE')._dirty_watches = set() # Watch UUIDs that need saving - client.application.config.get('DATASTORE')._dirty_settings = False # Settings changed - client.application.config.get('DATASTORE')._watch_hashes = {} # UUID -> SHA256 hash for change detection - uuids = list(client.application.config.get('DATASTORE').data['watching']) for uuid in uuids: client.application.config.get('DATASTORE').delete(uuid)