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.
turt2live
left a comment
turt2live
left a comment
There was a problem hiding this comment.
generally this seems fine - thanks for taking a look! I haven't checked it against the MSC yet, but more interested in ensuring it works within a context of the project structure first.
| 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.
| 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).
| 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. |
3 participants
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.