Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LarsJurgensen wants to merge 8 commits into
base: main
Choose a base branch
from
+193 −1,119
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
6006bb4
feat: route data-path uploads through the coordinator
f-firas Aug 14, 2026
f0ee164
feat: serve the tus protocol from the coordinator
f-firas Aug 14, 2026
77f9cff
feat: consume postprocessing results in the coordinator
f-firas Aug 14, 2026
86133fa
feat: WIP
f-firas Aug 17, 2026
6b608d1
feat: handle missing node id in tus
LarsJurgensen Aug 19, 2026
61dc4ab
feat: add changelog
LarsJurgensen Aug 20, 2026
3dc914a
feat: make TouchFile child-link creation atomic
LarsJurgensen Aug 20, 2026
79bb052
feat: add NewCoordinatorFromConfig
LarsJurgensen Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 49 additions & 0 deletions changelog/unreleased/feat-upload-coordinator.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
Enhancement: Extract the upload state machine into a driver-agnostic coordinator

The upload state machine (TUS session management, postprocessing event loop,
antivirus integration, and restart safety) has been extracted from decomposedfs
into a new coordinator in `pkg/upload`. Every storage driver now inherits TUS
chunked uploads, postprocessing, and AV scanning without reimplementing any of
it.

Drivers integrate by implementing four new methods on the storage interface:

- `MarkProcessing` sets or clears a "processing" flag on a resource so readers
see a grayed-out placeholder while bytes are in flight. Drivers that do not
need concurrent-upload protection may implement this as a no-op.
- `PrepareUpload` is called after all bytes are received and before
postprocessing begins. Decomposedfs uses this to lock the node, snapshot the
previous version, and propagate the optimistic size change. Drivers with no
such requirements may return immediately.
- `CommitUpload` writes the staged bytes to the resource and receives
pre-computed checksums.
- `RollbackUpload` is the inverse of `PrepareUpload` and is called when
postprocessing fails or is aborted. Drivers that returned immediately from
`PrepareUpload` may return nil. The `RollbackInfo` struct carries the node
identity from the upload session rather than from live node metadata, so a
rollback can still release the quota of a node whose metadata has become
unreadable (e.g. because an ancestor was trashed mid-upload).

The coordinator owns the upload session files for the decomposedfs driver at
the same on-disk location as before (`<root>/uploads/`), so existing in-flight
uploads continue without interruption and no migration is required.

**Configuration:**

Both storageprovider and dataprovider gain an `upload_directory` config key that
sets the local directory where temporary upload session files and staged bytes are stored.
For decomposedfs this is optional; the coordinator falls back to `<root>/uploads/`
inside the driver's own root directory. For drivers that have no local filesystem
root, `upload_directory` must be set explicitly; otherwise the service fails to start.

The postprocessing consumer settings (`asyncfileuploads`, `consumer_group`,
`numconsumers`, `mount_id`) are read from the driver's own config block, the
same keys decomposedfs already uses. No new top-level config is introduced.

https://github.com/owncloud/reva/pull/702
https://github.com/owncloud/reva/pull/703
https://github.com/owncloud/reva/pull/714
https://github.com/owncloud/reva/pull/715
https://github.com/owncloud/reva/pull/717
https://github.com/owncloud/reva/pull/720
https://github.com/owncloud/reva/pull/721
28 changes: 20 additions & 8 deletions internal/grpc/services/storageprovider/storageprovider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ import (
"github.com/owncloud/reva/v2/pkg/storage"
"github.com/owncloud/reva/v2/pkg/storage/fs/registry"
"github.com/owncloud/reva/v2/pkg/storagespace"
"github.com/owncloud/reva/v2/pkg/upload"
"github.com/owncloud/reva/v2/pkg/utils"
"github.com/pkg/errors"
"github.com/rs/zerolog"
Expand All @@ -71,6 +72,7 @@ type config struct {
MountID string `mapstructure:"mount_id"`
UploadExpiration int64 `mapstructure:"upload_expiration" docs:"0;Duration for how long uploads will be valid."`
Events eventconfig `mapstructure:"events" docs:"0;Event stream configuration"`
UploadDirectory string `mapstructure:"upload_directory" docs:";Local directory for staging upload sessions. Overrides the driver's root. Required for drivers that have no local filesystem root."`
}

type eventconfig struct {
Expand Down Expand Up @@ -106,6 +108,7 @@ func (c *config) init() {
type Service struct {
conf *config
Storage storage.FS
Coordinator upload.Coordinator
dataServerURL *url.URL
availableXS []*provider.ResourceChecksumPriority
}
Expand Down Expand Up @@ -175,7 +178,14 @@ func New(m map[string]interface{}, ss *grpc.Server, log *zerolog.Logger) (rgrpc.

c.init()

fs, err := getFS(c, log)
// One stream for both the driver and the coordinator: a second one would open a
// second nats connection for the same events.
evstream, err := estreamFromConfig(c.Events)
if err != nil {
return nil, err
}

fs, err := getFS(c, evstream, log)
if err != nil {
return nil, err
}
Expand All @@ -202,9 +212,16 @@ func New(m map[string]interface{}, ss *grpc.Server, log *zerolog.Logger) (rgrpc.
return nil, err
}

// storageprovider only initiates uploads; the data path assembles chunks, so no chunking here.
coord, err := upload.NewCoordinatorFromConfig(c.UploadDirectory, c.Drivers[c.Driver], fs, evstream, log, false)
if err != nil {
return nil, fmt.Errorf("storageprovider: %w", err)
}

service := &Service{
conf: c,
Storage: fs,
Coordinator: coord,
dataServerURL: u,
availableXS: xsTypes,
}
Expand Down Expand Up @@ -427,7 +444,7 @@ func (s *Service) InitiateFileUpload(ctx context.Context, req *provider.Initiate
metadata["expires"] = strconv.Itoa(int(expirationTimestamp.Seconds))
}

uploadIDs, err := s.Storage.InitiateUpload(ctx, req.Ref, uploadLength, metadata)
uploadIDs, err := s.Coordinator.InitiateUpload(ctx, req.Ref, uploadLength, metadata)
if err != nil {
var st *rpc.Status
switch err.(type) {
Expand Down Expand Up @@ -1266,12 +1283,7 @@ func (s *Service) addMissingStorageProviderID(resourceID *provider.ResourceId, s
}
}

func getFS(c *config, log *zerolog.Logger) (storage.FS, error) {
evstream, err := estreamFromConfig(c.Events)
if err != nil {
return nil, err
}

func getFS(c *config, evstream events.Stream, log *zerolog.Logger) (storage.FS, error) {
if f, ok := registry.NewFuncs[c.Driver]; ok {
driverConf := c.Drivers[c.Driver]
driverConf["mount_id"] = c.MountID // pass the mount id to the driver
Expand Down
22 changes: 19 additions & 3 deletions internal/http/services/dataprovider/dataprovider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ import (
"github.com/owncloud/reva/v2/pkg/rhttp/router"
"github.com/owncloud/reva/v2/pkg/storage"
"github.com/owncloud/reva/v2/pkg/storage/fs/registry"
"github.com/owncloud/reva/v2/pkg/upload"
)

func init() {
Expand All @@ -51,6 +52,7 @@ type config struct {
NatsEnableTLS bool `mapstructure:"nats_enable_tls"`
NatsUsername string `mapstructure:"nats_username"`
NatsPassword string `mapstructure:"nats_password"`
UploadDirectory string `mapstructure:"upload_directory" docs:";Local directory for staging upload sessions. Overrides the driver's root. Required for drivers that have no local filesystem root."`
}

func (c *config) init() {
Expand Down Expand Up @@ -104,7 +106,21 @@ func New(m map[string]interface{}, log *zerolog.Logger) (global.Service, error)
return nil, err
}

dataTXs, err := getDataTXs(conf, fs, evstream, log)
// the data path assembles chunks, so enable chunking
coord, err := upload.NewCoordinatorFromConfig(conf.UploadDirectory, conf.Drivers[conf.Driver], fs, evstream, log, true)
if err != nil {
return nil, fmt.Errorf("dataprovider: %w", err)
}

// only the data path consumes postprocessing results: one consumer group gets
// one copy of each event, so a second subscriber would take half of them
if ac := upload.AsyncConfFromDriverConf(conf.Drivers[conf.Driver]); ac.Enabled {
if err := coord.StartPostprocessing(evstream, ac.ConsumerGroup, ac.MountID, ac.NumConsumers); err != nil {

@LukasHirt LukasHirt Aug 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 LarsJurgensen Aug 21, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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{},
	}

return nil, fmt.Errorf("dataprovider: could not start postprocessing: %w", err)
}
}

dataTXs, err := getDataTXs(conf, coord, fs, evstream, log)
if err != nil {
return nil, err
}
Expand All @@ -126,7 +142,7 @@ func getFS(c *config, stream events.Stream, log *zerolog.Logger) (storage.FS, er
return nil, fmt.Errorf("driver not found: %s", c.Driver)
}

func getDataTXs(c *config, fs storage.FS, publisher events.Publisher, log *zerolog.Logger) (map[string]http.Handler, error) {
func getDataTXs(c *config, coord upload.Coordinator, fs storage.FS, publisher events.Publisher, log *zerolog.Logger) (map[string]http.Handler, error) {
if c.DataTXs == nil {
c.DataTXs = make(map[string]map[string]interface{})
}
Expand All @@ -146,7 +162,7 @@ func getDataTXs(c *config, fs storage.FS, publisher events.Publisher, log *zerol
for t := range c.DataTXs {
if f, ok := datatxregistry.NewFuncs[t]; ok {
if tx, err := f(c.DataTXs[t], publisher, log); err == nil {
if handler, err := tx.Handler(fs); err == nil {
if handler, err := tx.Handler(coord, fs); err == nil {
txs[t] = handler
}
}
Expand Down
3 changes: 2 additions & 1 deletion internal/http/services/owncloud/ocdav/tus.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,7 +319,8 @@ func (s *svc) handleTusPost(ctx context.Context, w http.ResponseWriter, r *http.
sReq.Ref.Path = uReq.Ref.GetPath()
sReq.Ref.ResourceId = nil
} else {
if resid, err := storagespace.ParseID(httpRes.Header.Get(net.HeaderOCFileID)); err == nil {
// new files have no node id yet; keep the path-based ref instead
if resid, err := storagespace.ParseID(httpRes.Header.Get(net.HeaderOCFileID)); err == nil && resid.GetOpaqueId() != "" {
sReq.Ref = &provider.Reference{
ResourceId: &resid,
}
Expand Down
5 changes: 4 additions & 1 deletion pkg/rhttp/datatx/datatx.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,12 +28,15 @@ import (
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
"github.com/owncloud/reva/v2/pkg/events"
"github.com/owncloud/reva/v2/pkg/storage"
"github.com/owncloud/reva/v2/pkg/upload"
"github.com/owncloud/reva/v2/pkg/utils"
)

// DataTX provides an abstraction around various data transfer protocols.
type DataTX interface {
Handler(fs storage.FS) (http.Handler, error)
// Handler serves the protocol's data path. Uploads go through coord, which
// owns the upload lifecycle for every driver; downloads read from driver.
Handler(coord upload.Coordinator, driver storage.FS) (http.Handler, error)
}

// EmitFileUploadedEvent is a helper function which publishes a FileUploaded event
Expand Down
9 changes: 5 additions & 4 deletions pkg/rhttp/datatx/manager/simple/simple.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,8 @@ import (

userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
ctxpkg "github.com/owncloud/reva/v2/pkg/ctx"
"github.com/mitchellh/mapstructure"
ctxpkg "github.com/owncloud/reva/v2/pkg/ctx"
"github.com/pkg/errors"
"github.com/rs/zerolog"

Expand All @@ -40,6 +40,7 @@ import (
"github.com/owncloud/reva/v2/pkg/storage"
"github.com/owncloud/reva/v2/pkg/storage/cache"
"github.com/owncloud/reva/v2/pkg/storagespace"
"github.com/owncloud/reva/v2/pkg/upload"
"github.com/owncloud/reva/v2/pkg/utils"
)

Expand Down Expand Up @@ -78,7 +79,7 @@ func New(m map[string]interface{}, publisher events.Publisher, log *zerolog.Logg
}, nil
}

func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
func (m *manager) Handler(coord upload.Coordinator, driver storage.FS) (http.Handler, error) {
h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
sublog := m.log.With().Str("path", r.URL.Path).Logger()
r = r.WithContext(appctx.WithLogger(r.Context(), &sublog))
Expand All @@ -92,7 +93,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
metrics.DownloadsActive.Sub(1)
}()
}
download.GetOrHeadFile(w, r, fs, "")
download.GetOrHeadFile(w, r, driver, "")
case "PUT":
metrics.UploadsActive.Add(1)
defer func() {
Expand All @@ -114,7 +115,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
ctx = ctxpkg.ContextSetLockID(ctx, lockID)
}

info, err := fs.Upload(ctx, storage.UploadRequest{
info, err := coord.Upload(ctx, storage.UploadRequest{
Ref: ref,
Body: r.Body,
Length: r.ContentLength,
Expand Down
7 changes: 4 additions & 3 deletions pkg/rhttp/datatx/manager/spaces/spaces.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ import (
"github.com/owncloud/reva/v2/pkg/storage"
"github.com/owncloud/reva/v2/pkg/storage/cache"
"github.com/owncloud/reva/v2/pkg/storagespace"
"github.com/owncloud/reva/v2/pkg/upload"
"github.com/owncloud/reva/v2/pkg/utils"
)

Expand Down Expand Up @@ -80,7 +81,7 @@ func New(m map[string]interface{}, publisher events.Publisher, log *zerolog.Logg
}, nil
}

func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
func (m *manager) Handler(coord upload.Coordinator, driver storage.FS) (http.Handler, error) {
h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var spaceID string
spaceID, r.URL.Path = router.ShiftPath(r.URL.Path)
Expand All @@ -97,7 +98,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
metrics.DownloadsActive.Sub(1)
}()
}
download.GetOrHeadFile(w, r, fs, spaceID)
download.GetOrHeadFile(w, r, driver, spaceID)
case "PUT":
metrics.UploadsActive.Add(1)
defer func() {
Expand All @@ -117,7 +118,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
Path: fn,
}
var info *provider.ResourceInfo
info, err = fs.Upload(ctx, storage.UploadRequest{
info, err = coord.Upload(ctx, storage.UploadRequest{
Ref: ref,
Body: r.Body,
Length: r.ContentLength,
Expand Down
Loading