butonic
commented
1 week ago The ocis charts always uses "OCIS_ASYNC_UPLOADS=true", so postprocessing is always enabled. The postprocessing code in the decomposedfs is handling the events:
1. case events.PostprocessingFinished:
by default deletes the postprocessing status.
only when the ev.Outcome is events.PPOutcomeContinue and the session.Finalize() fails will we not delete the postprocessing state
The Finalize() call tries to upload the blob. Only that can fail.
2. case events.RestartPostprocessing:
does not change postprocessing state
3. case events.PostprocessingStepFinished:
does not change postprocessing state
There are other codepaths that call session.store.Cleanup(), but they are all deleting the postprocessing state:
The FinishUpload() handling might call session.store.Cleanup():
1. on a checksum error
on a calculation error or an unknown checksum we call session.store.Cleanup(ctx, session, true, false, false). The signature of that function is Cleanup(ctx context.Context, session Session, revertNodeMetadata, keepUpload, unmarkPostprocessing bool) and it calls session.Cleanup(revertNodeMetadata, !keepUpload, !keepUpload). That signature is Cleanup(revertNodeMetadata, cleanBin, cleanInfo bool). When revertNodeMetadata is true and there is a node for the session and this is the first revision of a file we completely delete the node:
// cleanup cleans up after the upload is finished
func (session *OcisSession) Cleanup(revertNodeMetadata, cleanBin, cleanInfo bool) {
ctx := session.Context(context.Background())
if revertNodeMetadata {
n, err := session.Node(ctx)
if err != nil {
appctx.GetLogger(ctx).Error().Err(err).Str("sessionid", session.ID()).Msg("reading node for session failed")
} else {
if session.NodeExists() && session.info.MetaData["versionsPath"] != "" {
p := session.info.MetaData["versionsPath"]
if err := session.store.lu.CopyMetadata(ctx, p, n.InternalPath(), func(attributeName string, value []byte) (newValue []byte, copy bool) {
return value, strings.HasPrefix(attributeName, prefixes.ChecksumPrefix) ||
attributeName == prefixes.TypeAttr ||
attributeName == prefixes.BlobIDAttr ||
attributeName == prefixes.BlobsizeAttr ||
attributeName == prefixes.MTimeAttr
}, true); err != nil {
appctx.GetLogger(ctx).Info().Str("versionpath", p).Str("nodepath", n.InternalPath()).Err(err).Msg("renaming version node failed")
}
if err := os.RemoveAll(p); err != nil {
appctx.GetLogger(ctx).Info().Str("versionpath", p).Str("nodepath", n.InternalPath()).Err(err).Msg("error removing version")
}
} else {
// if no other upload session is in progress (processing id != session id) or has finished (processing id == "")
latestSession, err := n.ProcessingID(ctx)
if err != nil {
appctx.GetLogger(ctx).Error().Err(err).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("uploadid", session.ID()).Msg("reading processingid for session failed")
}
if latestSession == session.ID() {
// actually delete the node
session.removeNode(ctx)
}
// FIXME else if the upload has become a revision, delete the revision, or if it is the last one, delete the node
}
}
}
if cleanBin {
if err := os.Remove(session.binPath()); err != nil && !errors.Is(err, fs.ErrNotExist) {
appctx.GetLogger(ctx).Error().Str("path", session.binPath()).Err(err).Msg("removing upload failed")
}
}
if cleanInfo {
if err := session.Purge(ctx); err != nil && !errors.Is(err, fs.ErrNotExist) {
appctx.GetLogger(ctx).Error().Err(err).Str("session", session.ID()).Msg("removing upload info failed")
}
}
}
PrajwolAmatya
kobergj
rhafer
nicholas-wilson-au
We found a file that is stuck in postprocessing but the upload session has been deleted. I assume because the virus scanner emitted an event that caused the upload session to be deleted without removing the postprocessing flag.
maybe relatedy: https://github.com/owncloud/ocis/issues/8848 https://github.com/owncloud/ocis/issues/9244 https://github.com/owncloud/ocis/issues/7177