Enabling Distributed Processing

[markmac99]

An upgrade to the RMS solver, CorrelateRMS, to enable distributed processing across multiple servers. To assist with this the first step has been split into two, one to create candidates and one to perform initial simple solutions.

This PR also replaces the JSON database with SQLite which is necessary to enable distributed processing and also brings performance benefits. The PR adds two new commandline arguments, one to control how much data to retain in the databases, and one to post-fix the log name with the phase name eg correlate_rms_20260101_12345_cands.log. This is to ensure each phase's logfile is uniquely named and to make it easier to monitor and debug.

[markmac99]

1. Millisecond timestamp drift → duplicate trajectories (your known issue — root cause)

A trajectory's identity is its output folder, named from jdt_ref down to the millisecond (generateTrajOutputDirectoryPath, ~L1376), and the DB key is jdt_ref REAL UNIQUE. On a single machine this is handled: after a phase2 MC re-solve shifts jdt_ref, solveTrajectory (CorrelateEngine ~L1067-1071) notices and calls removeTrajectory(orig_traj, remove_phase1=True), which deletes the stale phase1 pickle, the old folder, and the old DB row.

Distributed, that last step can't happen: phase2 runs on a mode-2 child whose removeTrajectory path (CorrelateRMS L1466-1476) only deletes local folders and returns before touching the DB

In fact this should be handled already elsewhere.

Each trajectory contains a pre_mc_longname value, which identifies the phase1 solution. In moveUploadedData we read this field and if the folder exists, we delete it:

  traj = loadPickle(src_path, src_name)
  if hasattr(traj, 'pre_mc_longname') and traj.pre_mc_longname != traj.longname:
      # When saving MC-phase solns we need to check if the MC phase has changed the reference time
      # slightly, and if so, remove the original folder to ensure no duplicates.
      pretraj = traj.pre_mc_longname
      pre_path = os.path.join(self.output_dir, 'trajectories', pretraj[:4], pretraj[:6], pretraj[:8], pretraj)
      log.info(f'removing {pre_path}')
      if os.path.isdir(pre_path):
          shutil.rmtree(pre_path, ignore_errors=True)

Then, in updateTrajectoryDatabase, at around line 1100, we scan for trajectories with exactly the same observations, and remove whichever one isn't on disk from the database. We do remove by Traj ID, so there's a chance we'll remove both the ph1 and ph2 solns from the DB if the traj_id didn't change, but a few lines later at line 1173, we will re-add any trajs that are still on disk, so this will re-add the ph2 solns if they got inadvertently removed.

So, while its a bit convoluted, i think it does do what we need. and in testing, I did not encounter any situations where phase1 solns were orphaned. I will double check though.

[markmac99]

2. Files are published under their final names while the other side is reading them

Both directions write directly to the final filename and the other side polls for it, so a reader can pick up a half-written file:

• Parent → child: the parent writes candidate pickles with a plain open()+dump / shutil.copy into candidates/, while the child SFTP-lists that same folder.
• Child → parent: children upload results to final names while the parent copy+removes them and merges the uploaded .db files.

Note that this risk already exists because even on a single server the phase1 process could be writing to the phase1/ folder while the mc-phase process is reading from it.

Issues can arise in three places .

1. parent is distributing candidate or phase1 pickles to child node folders, while child is simultaneously downloading.
2. parent is scanning for uploads and processing files, while child is simultaneously uploading
3. parent phase1 process is saving phase1 solutions while the parent mc-phase is reading them.

To ensure partly-created pickles are not read by sftp or other processes on the parent, I've altered savePickle to use a unique temporary name and then rename the file afterwards.

To ensure partly-complete uploads are not read, i've altered putWithRetry to use a unique temporary name and then rename afterwards.

I've also created a function safeCopyOrMove() in OSTools, which uses a temporary name when copying or movnig files around, and i've used it wherever there was a chance of a read/write conflict. Note that we can't directly use os.replace() as this doesn't work across filesystems.

[markmac99]

I've made what i hope are the final changes. Candidates and phase1 solutions that are being processed are now no longer moved to the processed/ folders till after processing is complete. To prevent the same data being picked up by another process or remote node, the status is marked as 2 in the database. This entailed some changes to the DB class to handle concurrency better.

Testing:
I ran three instances of the solver on my server, one in each mode, plus two remote nodes one in Phase1 mode and one in MC mode. Using 25 days of data from 13 cameras, i ran the solvers continuously, introducing a new day's data at random intervals between 10 and 30 minutes apart. The test ran over an 18-hour window without issue.

The parent node distributed and collected data as expected as well as processing is own workload. The remote nodes collected data, processed and uploaded as expected. There was occasionally an upload failure sending data to the parent node but this is handled by sending the data on the next pass.

Let me know if you spot anything else i need to look at.