Asynchronous uploads (MSC2246) #364
Conversation
Expands async upload notify support to multiple processes. The logic is all instances of media repo subscribe to all shards for upload notify events and any instance published an event to one of them. This could be done in reverse as well but I found this somewhat easier to do.
upload: handle as PUT request According to matrix-org/matrix-spec-proposals#2246, the new upload endpoint is a PUT endpoint, not a POST.
| @@ -97,3 +130,21 @@ func (c *RedisCache) GetBytes(ctx rcontext.RequestContext, key string) ([]byte, | |||
| b, err := r.Bytes() | |||
| return b, err | |||
| } | |||
|
|
|||
| func (c *RedisCache) NotifyUpload(ctx rcontext.RequestContext, origin string, mediaId string) error { | |||
There was a problem hiding this comment.
Removing the redis notifications would be good, as it's not currently safe to be running the media repo as multiple processes. The approach for this is expected to look a bit different architecturally in the end.
| @@ -59,12 +100,16 @@ func UploadMedia(r *http.Request, rctx rcontext.RequestContext, user api.UserInf | |||
|
|
|||
| contentLength := upload_controller.EstimateContentLength(r.ContentLength, r.Header.Get("Content-Length")) | |||
|
|
|||
| media, err := upload_controller.UploadMedia(r.Body, contentLength, contentType, filename, user.UserId, r.Host, rctx) | |||
| media, err := upload_controller.UploadMedia(r.Body, contentLength, contentType, filename, user.UserId, r.Host, mediaId, rctx) | |||
There was a problem hiding this comment.
StoreDirect should probably be used instead of UploadMedia when the media ID is known, similar to an import. Though, the upload size check might need to be moved up to somewhere. Possibly a new function which calls StoreDirect but is named OverwriteExisting or something?
| UnusedExpiresAt int64 `json:"unused_expires_at"` | ||
| } | ||
|
|
||
| func CreateMedia(r *http.Request, rctx rcontext.RequestContext, user api.UserInfo) interface{} { |
There was a problem hiding this comment.
This isn't r0, so should be in the unstable directory instead (alongside the media info and local copy endpoints).
| @@ -48,21 +48,38 @@ func DownloadMedia(r *http.Request, rctx rcontext.RequestContext, user api.UserI | |||
| downloadRemote = parsedFlag | |||
| } | |||
|
|
|||
| var asyncWaitMs *int = nil | |||
There was a problem hiding this comment.
ideally this is a non-pointer int throughout, for code clarity more than anything. A wait time of zero is still semantically possible.
There was a problem hiding this comment.
Having it as a nillable pointer was to make it distinct it's not set vs it being set to zero. Not setting it from the request makes the code pick up the default from the configuration file.
|
Thanks again for the PR! I've been working on what is best described as a rewrite of the whole project and added MSC2246 support as a first-class citizen. That progress is tracked here: #407 It's a bit different from how you've implemented it here, but should have the same functionality. |
Initial implementation of MSC2246 after it was simplified.
With Redis signaling it's also possible to run matrix-media-repo behind a load balancer and get notified of finished uploads across processes.
Gated and prefixed as unstable as per the MSC.