The uploads sessions command builds a posix driver with async file
uploads enabled, which makes it join the "dcfs" JetStream queue group.
Clean/restart/resume events published by the command itself can then be
routed to the CLI's own short-lived consumer; the nats client auto-acks
them and they are lost when the process exits, leaving stale upload
sessions behind.
Disable asyncfileuploads for the CLI driver instance so it only lists
sessions and publishes events. Also emit [] instead of null when the
session list is empty so the JSON output stays parseable.
Ported from owncloud/ocis#11693. The uploads clean command now emits an events.CleanUpload event via the event stream rather than calling Purge directly. Requires the corresponding reva bump (vendor changes pulled in separately).
This allows to pass the multi-tenant enablement flag as a shared config
option to the reva service. This needs to be done for all reva services
since it ends up in a global variable in reva that is only initialized
once, by the service that is the first to parse its config.
See https://github.com/opencloud-eu/opencloud/issues/1563 for details.
The STORAGE_USERS_DATA_GATEWAY_URL env var is used in the `tokens` `datagateway_endpoint` reva configuration. That `DataGatewayEndpoint` is used by the decomposedfs driver to create urls:
```go
// URL returns a url to download an upload
func (session *DecomposedFsSession) URL(_ context.Context) (string, error) {
// [ ... ]
return joinurl(session.store.tknopts.DataGatewayEndpoint, tkn), nil
}
```
As the comment points out this URL is internally used when emitting events. Either in:
```go
func (session *DecomposedFsSession) FinishUploadDecomposed(ctx context.Context) error {
// [ ... ]
s, err := session.URL(ctx)
if err != nil {
return err
}
var iu *userpb.User
if utils.ExistsInOpaque(u.Opaque, "impersonating-user") {
iu = &userpb.User{}
if err := utils.ReadJSONFromOpaque(u.Opaque, "impersonating-user", iu); err != nil {
return err
}
}
if err := events.Publish(ctx, session.store.pub, events.BytesReceived{
UploadID: session.ID(),
URL: s,
```
or in
```go
// Postprocessing starts the postprocessing result collector
func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) {
// [ ... ]
s, err := session.URL(ctx)
if err != nil {
sublog.Error().Err(err).Msg("could not create url")
continue
}
metrics.UploadSessionsRestarted.Inc()
// restart postprocessing
if err := events.Publish(ctx, fs.stream, events.BytesReceived{
UploadID: session.ID(),
URL: s,
```
So, we do not need to go throught the proxy here.
Florian Schade