diff --git a/grib_processor.py b/grib_processor.py
index d7ef169080fccd66e99795bff0c47d3013febc6c..b75c21b84992a1b57f6d2c215882f930cc2c1834 100644
--- a/grib_processor.py
+++ b/grib_processor.py
@@ -625,18 +625,20 @@ def main() -> int | None:
             exchange='model'
         )
         LOG.info('Connected to %s', ', '.join(args.servers))
+
+        # maybe add loading/storing cache to disk after exits?
+        cache = defaultdict(_cache_entry)
+        LOG.debug('Starting cache cleaner')
+        threading.Thread(name='grib_pipeline_cleaner', target=_cache_cleaner, args=(cache, args.clean_interval, args.cache_ttl), daemon=True).start()
     except KeyboardInterrupt:
+        mq.disconnect()
         return
 
-    # maybe add loading/storing cache to disk after exits?
-    cache = defaultdict(_cache_entry)
-    LOG.debug('Starting cache cleaner')
-    threading.Thread(name='grib_pipeline_cleaner', target=_cache_cleaner, args=(cache, args.clean_interval, args.cache_ttl), daemon=True).start()
-
     try:
         gribs_from_dir(args.grib_dir, cache, args.watch_debounce)
     except KeyboardInterrupt:
         LOG.critical('Got interrupt, goodbye!')
+        mq.disconnect()
 
 
 if __name__ == "__main__":