DM-53501: Investigate Null values in camera_hexapod_compensation_offset_* column

[rcboufleur]

Nulls in the camera_hexapod_compensation_offset_* columns came from two places. First, the upsert was overwriting good data with nulls whenever a topic returned no results for a given exposure, the incoming row had a null in that column, and the old ON CONFLICT ... SET blindly replaced the stored value. Second, exposures that straddled the task boundary were being processed with an incomplete time window, producing partial computations. The upsert now uses COALESCE(excluded.column, table.column) to keep the existing value when the incoming one is null, and the processing loop only includes exposures and visits whose timespan falls entirely inside the task window.

A few supporting cleanups are included: DAO objects are cached per Transform instance instead of reconstructed on every run, the database pool uses QueuePool with pool_pre_ping instead of NullPool, and SAWarning output is routed to the structured logger. Latent bugs fixed along the way: task_update_counts was writing "failed" instead of "running", select_failed's butler_repo filter was dead code, and current_utc could raise NameError depending on which branch executed.

[bbrondel]

I have concerns about exposures getting discarded if they straddle an EFD query window. Please write up a Jira ticket to make sure that will get addressed and write your ticket number into an appropriate place in a code comment. Other than that, approved with the following comments.

[bbrondel]

Move lines 92-99 above the initialization of self._engines to get the intended exception message you had in mind.

[bbrondel]

There should probably be a check that all URIs in the list have the same dialect.

[bbrondel]

The connect_timeout arg doesn't work with sqlite. Do something like this instead:

if sgbd == "sqlite":
    connect_args = {"timeout": 5 }
    self.dialect = sqlite
elif sgbd == "postgresql":
    connect_args = {"connect_timeout": 5 }
    self.dialect = postgresql
else:
    raise ...

[rcboufleur]

@bbrondel
About the concerns of discarting exposures/visits, the timewindow is meant to cover this case. In process_task(), exposures/visits are queried with an expanded window (start - timewindow, end + timewindow), with timewindow currently set to 1 minute. Then _process_interval() discards only records that are not fully contained in that expanded window. Since exposures are 15s and visits are 30s, anything straddling the original task boundary should still fit inside the current buffer. Records crossing the expanded boundary are skipped to avoid computing values from incomplete EFD data. We can increase timewindow if needed, but we keep it as small as possible to avoid unnecessary load on InfluxDB.