add BFT enforcement for bulk storage objects

[isegall-da]

Fixes #6495

Pull Request Checklist

Cluster Testing

• If a cluster test is required, comment /cluster_test on this PR to request it, and ping someone with access to the DA-internal system to approve it.
• If an upgrade test is required, comment /upgrade_test on this PR to request it, and ping someone with access to the DA-internal system to approve it.
• If a hard-migration test is required (from the latest release), comment /hdm_test on this PR to request it, and ping someone with access to the DA-internal system to approve it.
• If a logical synchronizer upgrade test is required (from canton-3.5), comment /lsu_test on this PR to request it, and ping someone with access to the DA-internal system to approve it.

PR Guidelines

• Include any change that might be observable by our partners or affect their deployment in the release notes.
• Specify fixed issues with Fixes #n, and mention issues worked on using #n
• Include a screenshot for frontend-related PRs - see README or use your favorite screenshot tool

Merge Guidelines

• Make the git commit message look sensible when squash-merging on GitHub (most likely: just copy your PR description).

[isegall-da]

This was leaking pipelines, it created a new one that did not get destroyed when the main one was.

[ray-roestenburg-da]

Suggested change

	**Under Development, do not use in production yet** Get checksums for bulk history objects. Uses both staging and committed objects.
	**Under Development, do not use in production yet** Get checksums for bulk history objects. Searches for object_keys in both staging and committed objects.

Minor suggestion

[ray-roestenburg-da]

This doesn't really close, it creates a SyncCloseable (could be confusing)

[ray-roestenburg-da]

the mutex.exclusive looks like a special mutex to especially not having to use blocking:

 Lock to be used instead of blocking synchronized

so I think the blocking should be removed.

[ray-roestenburg-da]

I'd make this a class that owns one connection instead of a trait.

As a trait, the connection state is mixed invisibly into the host class, and the dependencies are method parameters, but only the first caller's arguments are ever used, so a second call site passing a different store or config silently gets the wrong connection.

The mixer also has to remember to wire closeScanConnection() into its closing sequence.

Suggestion:

  class PeerBftScanConnection(store: ScanStore, svName: String, ...) extends AutoCloseable {
    @volatile private var started = false
    lazy val connection: Future[BftScanConnection] = {
      started = true
      BftScanConnection.peerScanConnection(...)
    }
    override def close(): Unit = if (started) connection.foreach(_.close())
  }

Constructor params pin the dependencies, lazy val gives thread-safe create-once semantics (no mutex, no var, no blocking), caching the future avoids creating two connections on concurrent first calls, and whoever constructs it closes it. Both return a closed connection after close(). But here that's a local owner bug (use-after-close), not a silent failure when calling getOrCreateScanConnection.

If we ever want this fully airtight (post-close access fails loudly with AbortedDueToShutdown, and close() waits for an in-flight creation instead of racing it), extending FlagCloseable and wrapping the accessor in synchronizeWithClosingF fixes that. For a connection closed once at node shutdown I don't think we need that here though.

[ray-roestenburg-da]

Looks good overall, please have a look at my comments re: HasPeerBftScanConnection