Documentation
¶
Overview ¶
Package resumable 提供「分片上传/断点续传」的会话生命周期状态机原语： 会话创建（含秒传钩子）、分片登记、断点查询、完成、中止、孤儿清扫， 状态面纯内存、存储后端接口化（分片字节落哪由 ChunkStore 决定）， stdlib-only（pkg 零第三方依赖红线）。
为什么需要它 ¶
下游 zeus 两个仓**各自独立**手搓了整套同构状态机（media 主实现 ~700 行 + gin-vue-admin 又一份 ~400 行，pkg#361）：会话建立/分片幂等/断点对账/ 合并抢占/孤儿清扫每个上传服务都要面对，而并发正确性的坑——CAS 进度、 合并抢占、missing 对账、孤儿清扫——是必须单点的语义，两份实现的防御 注释里各自都踩过这四个坑。业务差异其实只在**去重键**（DedupResolver） 与**合并落存储**（MergeFunc）两处，其余全是共同面。
状态机 ¶
StatusUploading --ClaimForMerge--> StatusMerging --Finish--> StatusDone StatusUploading --Abandon--> StatusAborted StatusMerging --ReleaseClaim--> StatusUploading（校验/合并失败回滚）
四类并发坑的单点兑现（SessionStore 契约 + Manager 编排）：
- **CAS 进度**：Manager.PutChunk 幂等分片 + 写后存活确认 + AppendReceived 原子登记（只在收片态、重复登记幂等）；
- **合并抢占**：Manager.Complete 的 ClaimForMerge 是 0→1 原子迁移， 并发 Complete 恰好一个赢家进合并、输家拿 ErrMergeInProgress；
- **断点对账**：Manager.ListParts 双源取交（「已登记 ∧ 字节存活 ∧ 字节数吻合」才算已收）——登记了但字节丢/残缺的分片必须回缺片清单， 否则断点查询会把客户端骗去 Complete 才炸；
- **孤儿清扫**：Manager.Sweep 单点收口过期会话/无主分片/合并卡死 （过期判定走 UpdatedAt——收片与状态迁移都会续期）。
用法 ¶
sessions := resumable.NewMemSessionStore() // 状态面：纯内存
chunks := store{...} // 存储面：实现 ChunkStore
m := resumable.NewManager(sessions, chunks)
s, uploaded, err := m.Init(ctx, resumable.InitReq{
Owner: uid, Hash: h, Size: size, ChunkSize: 8 << 20,
}, func(ctx context.Context, hash string) (bool, error) {
return objectExists(ctx, hash) // 秒传：内容已存在
})
// uploaded=true ⇒ 秒传命中，直接复用既有对象
err = m.PutChunk(ctx, s.ID, idx, r, 8<<20) // 幂等分片 + 进度 CAS
recv, missing, st, _ := m.ListParts(ctx, s.ID) // 断点查询（双清单对账）
err = m.Complete(ctx, s.ID, func(ctx context.Context, s *resumable.Session, r io.Reader) error {
return putObject(ctx, s.Hash, r) // 合并落存储（流式）
})
err = m.Abort(ctx, s.ID, uid) // 归属校验 + 清分片
n, err := m.Sweep(ctx, 24*time.Hour) // 孤儿清扫
边界（如实声明，不伪装全知） ¶
- **合并的幂等性**：merge 成功而 Finish 失败时会话留在合并态、由 Sweep 收口（不重跑 merge）；merge 失败则回滚回收片态、调用方重试 Complete 会**再跑一次** merge——落存储的写入要么幂等要么自持事务；
- **超长分片的字节**：PutChunk 流式不过缓冲，超长判定发生在读满 want+1 之后——存储里可能短暂留下残片（不登记、对账算 missing）， 重传覆盖或 Abort/Sweep 清掉；
- **here 式的业务语义不内置**：秒传的「已存在」判定、合并的目标布局 （单对象/分块对象/去重库）、留存策略（完成即删 vs 定期删）都是钩子/ 调用方的事；Sweep 按「过期即清」的单一口径走；
- **存储后端的一致性**：本包假定 ChunkStore 的 Put 是「以写代换」、 Stat 是最终一致的字节数；最终一致存储上「写后存活确认」可能把 读旧值判成残缺（ErrChunkSize）——分片幂等重传即可修复，不影响 状态机正确性。
Index ¶
- Variables
- type ChunkStore
- type DedupResolver
- type InitReq
- type Manager
- func (m *Manager) Abort(ctx context.Context, id string, owner string) error
- func (m *Manager) Complete(ctx context.Context, id string, merge MergeFunc) error
- func (m *Manager) Init(ctx context.Context, req InitReq, dedup DedupResolver) (*Session, bool, error)
- func (m *Manager) ListParts(ctx context.Context, id string) (received, missing []int, status Status, err error)
- func (m *Manager) PutChunk(ctx context.Context, id string, idx int, r io.Reader, maxChunk int64) error
- func (m *Manager) Sweep(ctx context.Context, olderThan time.Duration) (int, error)
- type MemChunkStore
- func (m *MemChunkStore) DeletePrefix(ctx context.Context, id string) error
- func (m *MemChunkStore) Open(ctx context.Context, id string, idx int) (io.ReadCloser, error)
- func (m *MemChunkStore) Put(ctx context.Context, id string, idx int, r io.Reader) error
- func (m *MemChunkStore) Stat(ctx context.Context, id string, idx int) (int64, error)
- type MemSessionStore
- func (m *MemSessionStore) Abandon(ctx context.Context, id string) error
- func (m *MemSessionStore) AppendReceived(ctx context.Context, id string, idx []int) error
- func (m *MemSessionStore) ClaimForMerge(ctx context.Context, id string) (bool, error)
- func (m *MemSessionStore) Create(ctx context.Context, s *Session) error
- func (m *MemSessionStore) Delete(ctx context.Context, id string) error
- func (m *MemSessionStore) Finish(ctx context.Context, id string) error
- func (m *MemSessionStore) Get(ctx context.Context, id string) (*Session, error)
- func (m *MemSessionStore) List(ctx context.Context) ([]*Session, error)
- func (m *MemSessionStore) ReleaseClaim(ctx context.Context, id string) error
- type MergeFunc
- type Session
- type SessionStore
- type Status
Constants ¶
This section is empty.
Variables ¶
var ( // ErrSessionNotFound 会话不存在。 ErrSessionNotFound = errors.New("resumable: 会话不存在") // ErrSessionExists 会话 id 已存在（Create 撞 id）。 ErrSessionExists = errors.New("resumable: 会话已存在") // ErrChunkNotFound 分片不存在（Stat/Open 未命中）。 ErrChunkNotFound = errors.New("resumable: 分片不存在") // ErrSessionDone 会话已完成——完成态不可再收片/合并/中止。 ErrSessionDone = errors.New("resumable: 会话已完成") // ErrSessionAborted 会话已中止。 ErrSessionAborted = errors.New("resumable: 会话已中止") // ErrMergeInProgress 合并进行中：合并抢占的输家、或对合并中的会话 // 收片/中止。输家可轮询 [Manager.ListParts] 的状态等赢家收场。 ErrMergeInProgress = errors.New("resumable: 合并进行中") // ErrNotOwner 归属不符：Abort 的 owner 与会话归属不一致。 ErrNotOwner = errors.New("resumable: 归属不符") // ErrBadChunkIndex 分片序号越界（不在 [0, NumChunks) 内）。 ErrBadChunkIndex = errors.New("resumable: 分片序号越界") // ErrChunkSize 分片字节数与会话数学不符（偏短/残缺）。 ErrChunkSize = errors.New("resumable: 分片字节数不符") // ErrChunkTooLarge 分片字节数超过上限（会话分片大小或调用方 maxChunk）。 ErrChunkTooLarge = errors.New("resumable: 分片字节数超过上限") // ErrMissingChunks 完成时分片不齐（收到的缺清单见包装上下文）。 ErrMissingChunks = errors.New("resumable: 分片不齐") // ErrBadInit 会话参数非法（Hash 空、Size 负、ChunkSize 非正等）。 ErrBadInit = errors.New("resumable: 会话参数非法") )
状态机与存储面的哨兵错误（errors.Is 可判）。
Functions ¶
This section is empty.
Types ¶
type ChunkStore ¶
type ChunkStore interface {
Put(ctx context.Context, id string, idx int, r io.Reader) error
Stat(ctx context.Context, id string, idx int) (int64, error)
Open(ctx context.Context, id string, idx int) (io.ReadCloser, error)
DeletePrefix(ctx context.Context, id string) error
}
ChunkStore 是分片字节的存储后端（接口化：本地盘、对象存储、内存皆可）。 实现契约：
- Put 以写代换（同 id+idx 整体覆盖，不追加）；
- Stat 回已写字节数，缺失报 ErrChunkNotFound（实现的读侧错误经 errors.Is 可判这个哨兵，其他错误原样上抛——「读不到」与「没有」 必须可分）；
- Open 回已写字节的读侧（缺失报 ErrChunkNotFound），调用方负责 Close；
- DeletePrefix 删掉该会话的全部分片，**幂等**（没有即成功）—— 孤儿清扫靠它收口。
type DedupResolver ¶
DedupResolver 是秒传钩子：内容摘要已存在（对象已完整落库）返回 true， Init 即回 (nil, true, nil)——不再建会话、不再收片。业务差异在去重键语义 （按 hash、按 hash+size、按租户隔离……），所以做成钩子而不是内置规则。
type InitReq ¶
type InitReq struct {
// ID 会话 id；空则由 [Manager.Init] 生成（crypto/rand 16 字节 hex）。
ID string
// Owner 归属键（必填语义由调用方定，Abort/Sweep 靠它分清「谁的会话」）。
Owner string
// Hash 全量内容摘要——秒传钩子的去重键（必填非空）。
Hash string
// Size 全量字节数（>= 0）。
Size int64
// ChunkSize 分片大小（> 0）。
ChunkSize int64
}
InitReq 建会话的请求。
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager 把状态机与两类存储串成上传会话面：会话创建、分片登记、断点 查询、完成、中止、孤儿清扫。构造后并发安全（互斥在两个 store 里）。
func NewManager ¶
func NewManager(sessions SessionStore, chunks ChunkStore) *Manager
NewManager 构造 Manager。任一 store 为 nil 立即 panic（仓规总则①： 构造期非法值 fail-fast——nil store 到第一个调用点才炸就晚了）。
func (*Manager) Abort ¶
Abort 中止会话（归属校验）：校验 owner 后把会话打到 StatusAborted 并清掉全部分片。StatusAborted 态再来是幂等收口（再清一次分片）； StatusDone 不可中止（ErrSessionDone）；合并中不可中止 （ErrMergeInProgress——等赢家收场再决定删不删）。
func (*Manager) Complete ¶
Complete 完成会话：合并抢占 → 齐全校验 → merge 交付 → 收尾清分片。
- 合并抢占（0→1 原子）：并发 Complete 恰好一个赢家进合并，输家拿 ErrMergeInProgress（可轮询等赢家收场）；StatusDone 态再来是 幂等 nil，StatusAborted 态报 ErrSessionAborted；
- 齐全校验：逐片「已登记 ∧ 字节存活 ∧ 字节数吻合」双源对账，缺一片 都不合并——回滚回收片态（ReleaseClaim）并报 ErrMissingChunks， 客户端补传后可直接重试（会话不换）；
- merge 失败同样回滚回收片态（瞬态故障重试不丢已传分片）；
- merge 成功即 Finish（StatusDone）并删分片（合并结果已交付业务 存储，分片留着就是孤儿字节）；Finish 失败时会话留在合并态由 Manager.Sweep 收口（merge 已交付，不重跑）。
func (*Manager) Init ¶
func (m *Manager) Init(ctx context.Context, req InitReq, dedup DedupResolver) (*Session, bool, error)
Init 建会话（或命中秒传）：
dedup 命中 → (nil, true, nil)：内容已存在，调用方直接复用既有对象； 否则按 req 建会话（StatusUploading）并返回 (会话, false, nil)。
req.ID 空则生成；Hash 空、Size<0、ChunkSize<=0 报 ErrBadInit。
func (*Manager) ListParts ¶
func (m *Manager) ListParts(ctx context.Context, id string) (received, missing []int, status Status, err error)
ListParts 断点查询（续传对账）：返回已收/缺片双清单与会话状态。
对账口径是**双源取交**：「已登记 ∧ 字节存活 ∧ 字节数吻合」才算已收—— 登记了但字节丢/残缺的分片必须回缺片清单（客户端重传），登记了却拿去 断点续传会在 Complete 才炸、把断点查询变成谎言。读不到（存储故障）不 归为缺片：错误原样上抛，不给客户端「重传能修好」的假答案。
func (*Manager) PutChunk ¶
func (m *Manager) PutChunk(ctx context.Context, id string, idx int, r io.Reader, maxChunk int64) error
PutChunk 登记一片（幂等分片 + CAS 进度 + 写后存活确认）：
- 只收 StatusUploading 态的会话（合并中/完成/中止各有各的哨兵）；
- idx 必须落在 [0, NumChunks)，否则 ErrBadChunkIndex；
- 字节数必须**恰好**是会话数学给的应有长度（非末片 = ChunkSize、 末片 = 余数）：超长报 ErrChunkTooLarge、偏短/残缺报 ErrChunkSize ——宁可让客户端重传一片，也不让「大小不对的分片」流进合并；
- maxChunk 是调用方防线（客户端声称的分片大小不可信时的护栏）： 低于本片应有长度直接报 ErrChunkTooLarge，读取上限取会话数学；
- 写后存活确认：Put 之后 Stat 复核字节数才登记进度——「写了」不等于 「写住了」，断点对账把「登记了但字节丢了」的分片算进 missing。
type MemChunkStore ¶
type MemChunkStore struct {
// contains filtered or unexported fields
}
MemChunkStore 是 ChunkStore 的内存实现（并发安全）。分片整块驻留 内存——流式上限由 Manager.PutChunk 的会话数学+maxChunk 把门， 直接拿本类型当裸存储用的调用方自限。同 id+idx 以写代换；Open 返回的 读器基于不可变字节块，写覆盖不干扰进行中的读。
func (*MemChunkStore) DeletePrefix ¶
func (m *MemChunkStore) DeletePrefix(ctx context.Context, id string) error
DeletePrefix 删掉会话全部分片；没有即成功（幂等——孤儿清扫靠它收口）。
func (*MemChunkStore) Open ¶
func (m *MemChunkStore) Open(ctx context.Context, id string, idx int) (io.ReadCloser, error)
Open 回分片读器；缺失报 ErrChunkNotFound。读器基于写入时的不可变 块（覆盖写生成新块，进行中的读不受影响）。
func (*MemChunkStore) Stat ¶
Stat 回分片字节数；缺失报 ErrChunkNotFound。
type MemSessionStore ¶
type MemSessionStore struct {
// contains filtered or unexported fields
}
MemSessionStore 是 SessionStore 的内存实现（并发安全）。 CAS 进度 / 合并抢占 / 状态单向 / 读侧快照四条契约都在互斥里单点兑现。
func (*MemSessionStore) Abandon ¶
func (m *MemSessionStore) Abandon(ctx context.Context, id string) error
Abandon 中止：非终态 → StatusAborted；StatusAborted 幂等； StatusDone 不可中止（报错）。
func (*MemSessionStore) AppendReceived ¶
AppendReceived 登记分片进度（CAS 面）：只在 StatusUploading 态收登记， 已登记序号幂等；其他状态按对应哨兵报错（登记与收片状态强绑定—— 合并中/完成后再登记就是状态机破口）。
func (*MemSessionStore) ClaimForMerge ¶
ClaimForMerge 原子抢占（0→1）：StatusUploading → StatusMerging 回 true；其余状态回 false（幂等读，不报错——输家的语义就是「没抢到」）。
func (*MemSessionStore) Create ¶
func (m *MemSessionStore) Create(ctx context.Context, s *Session) error
Create 登记新会话；id 重复报 ErrSessionExists。
func (*MemSessionStore) Delete ¶
func (m *MemSessionStore) Delete(ctx context.Context, id string) error
Delete 删会话记录（含进度）；缺失报 ErrSessionNotFound（清扫面把 「没删到」当真实失败，不吞）。
func (*MemSessionStore) Finish ¶
func (m *MemSessionStore) Finish(ctx context.Context, id string) error
Finish 收场：StatusMerging → StatusDone；其余状态报错。
func (*MemSessionStore) Get ¶
Get 回会话快照（深拷贝）；缺失报 ErrSessionNotFound。
func (*MemSessionStore) List ¶
func (m *MemSessionStore) List(ctx context.Context) ([]*Session, error)
List 回全部会话快照（按 id 升序，确定性输出）。
func (*MemSessionStore) ReleaseClaim ¶
func (m *MemSessionStore) ReleaseClaim(ctx context.Context, id string) error
ReleaseClaim 回滚合并抢占：StatusMerging → StatusUploading；其余状态 报错（回滚只对「抢了没交」的会话有意义）。
type MergeFunc ¶
MergeFunc 是合并落存储的钩子：r 为**全部分片按序号拼接**的字节流 （流式打开，不整对象进内存），把 r 写进业务存储即完成交付。合并成功后 Manager 负责 Finish 与分片回收；merge 返回 error 时会话回滚回收片态 （ReleaseClaim），调用方可修正后重试 Complete。业务差异只在落哪（对象 存储、文件、去重库……），并发正确性（谁来合并、缺片不合并、孤儿收口） 由状态机单点兜住。
type Session ¶
type Session struct {
// ID 会话 id（Init 生成或调用方给定）。
ID string
// Owner 归属键（调用方语义：用户 id、租户键……），Abort 按它校验。
Owner string
// Hash 全量内容摘要（去重键候选，秒传钩子的入参）。
Hash string
// Size 全量字节数（>= 0；0 表示空对象，无分片）。
Size int64
// ChunkSize 分片大小（> 0）；末片可短于它。
ChunkSize int64
// Status 当前状态。
Status Status
// CreatedAt / UpdatedAt 会话建立 / 最后进展时刻（Sweep 的过期判据）。
CreatedAt time.Time
UpdatedAt time.Time
// Received 是**已登记且存活确认过**的分片序号（升序快照）。
// 断点对账的双源之一：状态是进度快照，字节是真值（见 ListParts）。
Received []int
}
Session 是一个上传会话的事实面（SessionStore 的读侧快照——真值在 store，返回值是拷贝，改它不改 store）。
type SessionStore ¶
type SessionStore interface {
Create(ctx context.Context, s *Session) error
Get(ctx context.Context, id string) (*Session, error)
List(ctx context.Context) ([]*Session, error)
AppendReceived(ctx context.Context, id string, idx []int) error
ClaimForMerge(ctx context.Context, id string) (bool, error)
ReleaseClaim(ctx context.Context, id string) error
Finish(ctx context.Context, id string) error
Abandon(ctx context.Context, id string) error
Delete(ctx context.Context, id string) error
}
SessionStore 是会话状态机的存储后端。实现契约（并发正确性的四类坑都 压在这份接口上，实现必须单点兜住）：
- **CAS 进度**：AppendReceived 只在 StatusUploading 态登记、已登记 序号幂等；并发登记不丢、不重；
- **合并抢占**：ClaimForMerge 的 Uploading→[StatusMerging] 是原子 0→1 迁移，恰好一个赢家（输家拿 false）；
- **状态单向**：Finish 只从 Merging→[StatusDone]；ReleaseClaim 只从 Merging→Uploading（合并失败回滚，重试不换会话）；Abandon 把非终态 打到 StatusAborted（Aborted 幂等，Done 不可中止）；
- **读侧快照**：Get/List 返回深拷贝，调用方改写不污染内部状态。
type Status ¶
type Status int
Status 会话状态机的状态。状态迁移是单点语义（SessionStore 保证原子）：
StatusUploading --ClaimForMerge--> StatusMerging --Finish--> StatusDone StatusUploading --Abandon--> StatusAborted StatusMerging --ReleaseClaim--> StatusUploading（校验/合并失败回滚）
Source Files