mirror of
https://github.com/OpenListTeam/OpenList.git
synced 2026-10-10 21:13:10 +08:00
cca773f07b
* feat(multipart): add chunk reassembly window - Reassemble concurrently uploaded chunks into a sequential stream through a ring file, bounding disk usage to slots*chunkSize per session - Park writers up to a deadline when their slot is busy instead of rejecting instantly, so flow control does not surface as connection errors in browsers - Record a per-chunk CRC32 table for re-fill verification and keep it readable after close - Propagate cancellation to blocked readers via CloseWithError so drivers treat aborts like canceled requests - Cover ordering, backpressure, idempotent resends and close/abort wake-ups with race-enabled tests * feat(multipart): add pipelined upload session manager - Start the driver upload at session init over a sequential stream backed by the window, so client-to-server and server-to-storage transfers run concurrently - Attach client-provided hashes to the stream so drivers can attempt rapid upload before any chunk arrives, and absorb chunks racing pipeline completion idempotently - Keep only metadata and chunk CRCs after a failed attempt: re-sending chunk 0 re-fills a fresh window, and content changes between attempts are rejected - Resume receiving sessions only when client hashes prove the same file; failed_retriable sessions resume unconditionally - Reclaim sessions with a sliding-TTL GC and sweep orphaned ring files at startup - Cover the state machine with race-enabled tests over a stubbed storage layer * feat(setting): add multipart upload settings - Add multipart_enabled and multipart_chunk_size (MB) as public traffic settings - Validate the chunk size on save (integer within 1-90) via the setting item hook * feat(server): add multipart upload API - Add /api/fs/multipart init/chunk/complete/status/abort endpoints with headers aligned with /fs/put - Gate init behind the FsUp permission checks and reuse the client upload rate limiter for chunk uploads - Drain the request body before answering chunk requests on every path, so browsers do not see early responses as network errors - Start the multipart session GC when the router is initialized to reclaim ring files orphaned by a previous run * refactor(multipart): remove unused overwrite session field - The overwrite flag was stored on the session but never read: the handler performs the pre-check and op.Put owns the overwrite semantics * feat(setting): allow any positive multipart chunk size - Validate the setting as a positive integer only; self-hosted admins decide the ceiling themselves instead of an arbitrary 90MB cap - Keep clamping the client-suggested X-Chunk-Size to the admin value: the server buffers a window of 8 chunks per session, so an unbounded client suggestion would translate directly into server-side disk usage * refactor(server): simplify multipart chunk size clamp - Fold the two-step clamp into one branch: the ceiling is already floored, so a client suggestion just lowers the size with a 1MB floor
126 lines
3.5 KiB
Go
126 lines
3.5 KiB
Go
package op
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"regexp"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/OpenListTeam/OpenList/v4/internal/conf"
|
|
"github.com/OpenListTeam/OpenList/v4/internal/driver"
|
|
"github.com/OpenListTeam/OpenList/v4/internal/model"
|
|
"github.com/OpenListTeam/OpenList/v4/pkg/utils"
|
|
"github.com/pkg/errors"
|
|
log "github.com/sirupsen/logrus"
|
|
)
|
|
|
|
// Obj
|
|
type ObjsUpdateHook = func(ctx context.Context, parent string, objs []model.Obj)
|
|
|
|
var (
|
|
objsUpdateHooks = make([]ObjsUpdateHook, 0)
|
|
)
|
|
|
|
func RegisterObjsUpdateHook(hook ObjsUpdateHook) {
|
|
objsUpdateHooks = append(objsUpdateHooks, hook)
|
|
}
|
|
|
|
func HandleObjsUpdateHook(ctx context.Context, parent string, objs []model.Obj) {
|
|
for _, hook := range objsUpdateHooks {
|
|
hook(ctx, parent, objs)
|
|
}
|
|
}
|
|
|
|
// Setting
|
|
type SettingItemHook func(item *model.SettingItem) error
|
|
|
|
var settingItemHooks = map[string]SettingItemHook{
|
|
conf.VideoTypes: func(item *model.SettingItem) error {
|
|
conf.SlicesMap[conf.VideoTypes] = strings.Split(item.Value, ",")
|
|
return nil
|
|
},
|
|
conf.AudioTypes: func(item *model.SettingItem) error {
|
|
conf.SlicesMap[conf.AudioTypes] = strings.Split(item.Value, ",")
|
|
return nil
|
|
},
|
|
conf.ImageTypes: func(item *model.SettingItem) error {
|
|
conf.SlicesMap[conf.ImageTypes] = strings.Split(item.Value, ",")
|
|
return nil
|
|
},
|
|
conf.TextTypes: func(item *model.SettingItem) error {
|
|
conf.SlicesMap[conf.TextTypes] = strings.Split(item.Value, ",")
|
|
return nil
|
|
},
|
|
conf.ProxyTypes: func(item *model.SettingItem) error {
|
|
conf.SlicesMap[conf.ProxyTypes] = strings.Split(item.Value, ",")
|
|
return nil
|
|
},
|
|
conf.ProxyIgnoreHeaders: func(item *model.SettingItem) error {
|
|
conf.SlicesMap[conf.ProxyIgnoreHeaders] = strings.Split(item.Value, ",")
|
|
return nil
|
|
},
|
|
conf.PrivacyRegs: func(item *model.SettingItem) error {
|
|
regStrs := strings.Split(item.Value, "\n")
|
|
regs := make([]*regexp.Regexp, 0, len(regStrs))
|
|
for _, regStr := range regStrs {
|
|
reg, err := regexp.Compile(regStr)
|
|
if err != nil {
|
|
return errors.WithStack(err)
|
|
}
|
|
regs = append(regs, reg)
|
|
}
|
|
conf.PrivacyReg = regs
|
|
return nil
|
|
},
|
|
conf.FilenameCharMapping: func(item *model.SettingItem) error {
|
|
err := utils.Json.UnmarshalFromString(item.Value, &conf.FilenameCharMap)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
log.Debugf("filename char mapping: %+v", conf.FilenameCharMap)
|
|
return nil
|
|
},
|
|
conf.IgnoreDirectLinkParams: func(item *model.SettingItem) error {
|
|
conf.SlicesMap[conf.IgnoreDirectLinkParams] = strings.Split(item.Value, ",")
|
|
return nil
|
|
},
|
|
conf.MultipartChunkSize: func(item *model.SettingItem) error {
|
|
size, err := strconv.Atoi(strings.TrimSpace(item.Value))
|
|
if err != nil || size < 1 {
|
|
// deliberately a plain error: SaveSettings formats hook errors
|
|
// with %+v, which would dump a full stack trace into the UI
|
|
// notification for stack-carrying errors
|
|
return fmt.Errorf("multipart chunk size must be a positive integer (MB), got %q", item.Value)
|
|
}
|
|
item.Value = strconv.Itoa(size)
|
|
return nil
|
|
},
|
|
}
|
|
|
|
func RegisterSettingItemHook(key string, hook SettingItemHook) {
|
|
settingItemHooks[key] = hook
|
|
}
|
|
|
|
func HandleSettingItemHook(item *model.SettingItem) (hasHook bool, err error) {
|
|
if hook, ok := settingItemHooks[item.Key]; ok {
|
|
return true, hook(item)
|
|
}
|
|
return false, nil
|
|
}
|
|
|
|
// Storage
|
|
type StorageHook func(typ string, storage driver.Driver)
|
|
|
|
var storageHooks = make([]StorageHook, 0)
|
|
|
|
func callStorageHooks(typ string, storage driver.Driver) {
|
|
for _, hook := range storageHooks {
|
|
hook(typ, storage)
|
|
}
|
|
}
|
|
|
|
func RegisterStorageHook(hook StorageHook) {
|
|
storageHooks = append(storageHooks, hook)
|
|
}
|