feat: [OCISDEV-900] Connect storageprovider with coordinator instead of driver

[LarsJurgensen]

Wire storageprovider and dataprovider to the upload coordinator. This is the final step that makes TUS, postprocessing, AV scanning, and session management driver-independent.

• storageprovider.InitiateFileUpload routes through coord.InitiateUpload instead of fs.InitiateUpload
• Dataprovider starts the postprocessing NATS consumer on the coordinator, not on the driver
• datatx.DataTX.Handler gains a coord argument alongside the driver; all three implementations (simple, spaces, tus) updated
• PUT uploads go through coord.Upload; TUS registers coord.UseIn instead of the driver's UseIn
• ComposableFS check in the TUS manager removed; the coordinator satisfies the tusd data store interface for every driver
• Decomposedfs no longer subscribes to PostprocessingFinished, PostprocessingStepFinished, RestartPostprocessing, CleanUpload; the coordinator owns those
• Decomposedfs keeps RevertRevision under a separate consumer group (<group>-revisions) so it gets its own copy without contending with the coordinator's group
• Both services gain an upload_directory config key; required for drivers with no local filesystem root

[kw-security]

✅ Snyk checks have passed. No issues have been found so far.

Status	Scan Engine	Critical	High	Medium	Low	Total (0)
✅	Open Source Security	0	0	0	0	0 issues
✅	Licenses	0	0	0	0	0 issues
✅	Code Security	0	0	0	0	0 issues

💻 Catch issues earlier using the plugins for VS Code, JetBrains IDEs, Visual Studio, and Eclipse.

[LukasHirt]

Two correctness regressions, one reachability concern, two cleanups.

Blocking:

1. TouchFile's create-only symlink leaks an orphaned node on a concurrent-create race (pkg/storage/utils/decomposedfs/tree/tree.go:175).
2. OC-FileId is now permanently omitted from TUS responses for newly-created files (pkg/rhttp/datatx/manager/tus/tus.go:216).

Worth a look:
3. This PR is what makes pkg/upload's postprocessing path live for the dataprovider — and that path has no guard against a node vanishing mid-postprocessing (internal/http/services/dataprovider/dataprovider.go:117).

Minor:
4. Doubled NATS subscriptions for combined decomposedfs+dataprovider deployments (same line).
5. getCoordinator duplicated near-verbatim with storageprovider.go:234 (internal/http/services/dataprovider/dataprovider.go:146).

[LukasHirt]

The comment above claims the loser "gets AlreadyExists and its CAS loop re-reads and retries" — no such retry exists anywhere in the call chain (touchNode → Decomposedfs.TouchFile → here, each called exactly once, no re-read). The only CAS loop in the repo is in receivedsharecache.go for OCISDEV-855, an unrelated feature — looks like the comment got copied from there without the retry logic.

Net effect: on a real race, the node file/xattrs written above are permanently orphaned when the symlink loses, and neither session.Cleanup nor Decomposedfs.TouchFile's error path removes them.

[LarsJurgensen]

The only CAS loop in the repo is in receivedsharecache.go for OCISDEV-855, an unrelated feature

It's unrelated in a way. But the share service is also "uploading" files, so going through this logic. In the old upload flow, it did not return in error, but overwrite the file. In OCISDEV-855 this was fixed, to return an error and have a retry loop in share service. Now, the implementation lives in a different place, so we need the same (or very similar) change again. So two concurrent touchfile operations now make the second one fail & the cache service retries it. There are tests for this, which failed before this change, and pass now.

[LarsJurgensen]

no such retry exists anywhere in the call chain

This is the call chain:

retryPersist (20-attempt CAS loop)          receivedsharecache.go:232
    → persist → storage.Upload (IfNoneMatch=*) receivedsharecache.go:382
    → CS3.Upload → InitiateFileUpload           cs3.go:228
    → storageprovider.InitiateFileUpload        storageprovider.go:462
    → coordinator.touchNode                     coordinator.go:304
    → Decomposedfs.TouchFile                    decomposedfs.go:566
    → tree.TouchFile: os.Symlink → AlreadyExists tree.go:175-177

[LukasHirt]

info.Storage["NodeExists"] is only ever set once in populateSession, for whether the target pre-existed. touchNode never updates it after minting a new node ID. So this isn't "skipped for new files" as the comment says — it's permanently false for them, including on the completing response, so OC-FileId never reaches the client for a new upload.

[LarsJurgensen]

It is correct, that "NodeExists" represents, whether the node existed before the upload. This is equivalent to the previous flag in OcisSession. Other logic depends on this flag, so we should not change it.

Whether this check is correct is more complicated: In the old flow, when an upload is triggered, immediately a node id is generated. Then later in the flow when the node is actually created, decomposedfs created a node with that id. In the new implementation, this is no longer possible. We use TouchFile to create the file and it does not take a node id as parameter. We can't add it as a parameter, because some drivers might not be able to create a node given an id. Therefore, we only want to expose the id, once the node was actually created.

setHeaders runs before a PostFile / PatchFile, so before the node was actually created and thus before we can guarantee a stable id. After the upload is completed, it gets the actual id from client.Stat.

[LukasHirt]

onPostprocessingFinished (pkg/upload/postprocessing.go:139) has no equivalent to the deleted decomposedfs.go handler's guard for a node going missing mid-postprocessing — a failed CommitUpload just logs and calls publishUploadFailed, no session.Cleanup/RollbackUpload. Session + reserved quota get stuck with no auto-recovery. Not introduced by this diff, but this call site is what first makes it reachable for dataprovider in production.

Also: combined decomposedfs+dataprovider deployments now run two full-stream NATS subscriptions instead of one (decomposedfs's renamed <group>-revisions group plus this one) — each distinct group name gets a full copy of every event.

[LarsJurgensen]

a failed CommitUpload just logs and calls publishUploadFailed, no session.Cleanup/RollbackUpload

This is true, but intentional. The purpose is to allow followup actions from admin, i.e. CleanUpload or RestartPostprocessing. This is the same in old code:

case events.PPOutcomeContinue:
			if err := session.Finalize(ctx); err != nil {
				sublog.Error().Err(err).Msg("could not finalize upload")
				failed = true
				revertNodeMetadata = false
				keepUpload = true
				// keep postprocessing status so the upload is not deleted during housekeeping
				unmarkPostprocessing = false
			}

This basically leads to session.Cleanup(false, false, false, false), which is a noop. The upload session was kept & the processing flag is kept.

Also: combined decomposedfs+dataprovider deployments now run two full-stream NATS subscriptions instead of one (decomposedfs's renamed -revisions group plus this one) — each distinct group name gets a full copy of every event.

Previously, all events were handled in the driver decomposedfs. We want to make it driver independent, so handle them in the coordinator now (PostprocessingFinished, PostprocessingStepFinished, RestartPostprocessing, CleanUpload). However, there is one event, which really only makes sense for decomposedfs: RevertRevision. It's only triggered when a admin via CLI runs some cleanup script. This cleanup script is implemented very decomposedfs specific and only works for this driver. Then this script publishes this event. Having this in the coordinator is weird, because for every other driver it won't work.

Probably it would be nices to implement this somehow without using events at all, so decomposedfs would not need any event handling anymore. But for the purpose of implementing the coordinator, I don't want to reimplement the decomposedfs cleanup job. So for now the decomposedfs driver is still handling one event (RevertRevision) and all other events are handled by coordinator. To make sure it does not interfere with each other, the only register an independent subset of events:

var RegisteredEvents = []events.Unmarshaller{
	events.PostprocessingFinished{},
	events.PostprocessingStepFinished{},
	events.RestartPostprocessing{},
	events.CleanUpload{},
}

	_registeredEvents = []events.Unmarshaller{
		events.RevertRevision{},
	}

[LukasHirt]

Duplicated near-verbatim with storageprovider.go:234 — same NewFileStoreFromConfig → nil-check → Setup() → NewCoordinator sequence, differing only in the chunk-folder arg and error prefix. Worth a shared upload.NewCoordinatorFromConfig(...) helper so the two can't silently diverge.

[LarsJurgensen]

updated 👍