Abandoned Uploads - Upstash Documentation

A user picks a file, the upload starts, and the tab closes halfway through. No callback runs and your server never hears about it again. This page covers what that leaves in the bucket and how to clean it up.

What is left depends on which side of the multipart threshold the file was on:

	Under the threshold, one PUT	Over the threshold, multipart
When the object exists	the moment the last byte lands	when `end` completes the upload
What a dead tab leaves	a whole stored object, visible to `list()`	incomplete parts, invisible to `list()`
Who cleans it up	you, with a pending row or a list sweep	aborting stale uploads

One cron handles both halves. The multipart half is one call. The single-PUT half needs your app’s help, because the SDK cannot tell an abandoned object from a completed one by looking at the bucket. Only your own records can.

The simplest fix: always multipart#

// lib/uploads.tsexport const uploads = uploadHandler({  multipart: true,  onBeforeUpload: ({ file }) => ({ path: uniquePath(`uploads/${file.name}`) }),  onUploadComplete: async ({ path }) => recordFile(path),})

# app/uploads.pyasync def record(args: UploadCompleteArgs) -> None:    await record_file(args.path)uploads = AsyncUploadHandler(    multipart=True,    on_before_upload=lambda args: BeforeUploadResult(path=unique_path(f"uploads/{args.file.name}")),    on_upload_complete=record,)

// uploads.gouploads, err := blob.NewUploadHandler(blob.UploadHandlerOptions[struct{}]{	Multipart: blob.MultipartAlways,	OnBeforeUpload: func(		a blob.BeforeUploadArgs[struct{}],	) (blob.BeforeUploadResult, error) {		path := blob.UniquePath("uploads/" + a.File.Name)		return blob.BeforeUploadResult{Path: path}, nil	},	OnUploadComplete: func(a blob.UploadCompleteArgs[struct{}]) (any, error) {		return nil, recordFile(a.Request.Context(), a.Path)	},})

Set multipart on, on the handler or a route, and every upload uses parts, whatever its size. Nothing is stored until your handler completes the upload, so a closed tab always leaves an incomplete multipart upload, which the cron below cleans up. The single-PUT case disappears, and small files gain pause, resume and per-part retry.

The cost is two extra round trips between your server and storage per upload, and no extra browser requests. For most apps this one option is enough.

Sweeping incomplete multipart uploads#

// app/api/cron/sweep-uploads/route.tsimport { Bucket } from "@upstash/blob"const bucket = Bucket.fromEnv()export const GET = async () => {  const aborted = await bucket.abortStaleMultipartUploads({ olderThan: "1d", prefix: "uploads/" })  return Response.json({ aborted: aborted.length, paths: aborted.map((u) => u.path) })}

# app/cron.pyfrom upstash_blob import Bucketbucket = Bucket.from_env()def sweep_uploads() -> dict:    aborted = bucket.abort_stale_multipart_uploads(older_than="1d", prefix="uploads/")    return {"aborted": len(aborted), "paths": [u.path for u in aborted]}

// cron.gofunc sweepUploads(w http.ResponseWriter, r *http.Request) {	aborted, err := bucket.AbortStaleMultipartUploads(		r.Context(), 24*time.Hour, "uploads/",	)	if err != nil {		http.Error(w, err.Error(), http.StatusInternalServerError)		return	}	paths := make([]string, len(aborted))	for i, u := range aborted {		paths[i] = u.Path	}	json.NewEncoder(w).Encode(map[string]any{		"aborted": len(aborted),		"paths":   paths,	})}

vercel.json

{ "crons": [{ "path": "/api/cron/sweep-uploads", "schedule": "0 4 * * *" }] }

Aborting stale uploads lists the bucket’s incomplete uploads, aborts every one started longer ago than the older-than window, and returns what it aborted. Pick a window comfortably longer than your slowest upload, since a paused upload can be resumed long after it started. A day is a reasonable default.

On an all-multipart app this route is the whole cleanup. The calls it is built from are on Deleting.

Single PUT: the pending row#

If you keep the single-PUT path, write a row before the bytes, flip it after them, and sweep what never flipped from the same cron.

// lib/uploads.tsimport "server-only"import { BlobError, uniquePath, uploadHandler, uploadRoute } from "@upstash/blob"import { db } from "@/lib/db"const attachment = uploadRoute()({  constraints: { maxSize: "20mb", contentTypes: ["image/*", "application/pdf"] },  onBeforeUpload: async ({ request, file }) => {    const user = await getUser(request)    if (!user) throw new BlobError("unauthorized")    const rowId = crypto.randomUUID()    const path = uniquePath(`uploads/${user.id}/${file.name}`)    await db.uploads.insert({ id: rowId, owner: user.id, path, status: "pending" })    // state crosses in the completion token, so the callback can read it without a lookup.    return { path, state: { rowId } }  },  onUploadComplete: async ({ path, size, contentType, url, state }) => {    await indexForSearch(path, contentType)    await notifyOwner(state.rowId)    // Last. Everything above has to be done before the row stops looking abandoned.    await db.uploads.update(state.rowId, { status: "ready", size, url })    return { rowId: state.rowId }  },})export const uploads = uploadHandler({ routes: { attachment } })

# app/uploads.pyimport uuidfrom upstash_blob import AsyncUploadHandler, BeforeUploadResult, BlobError, UploadRoute, unique_pathfrom app.db import dbasync def begin_attachment(args):    user = await get_user(args.request)    if user is None:        raise BlobError("unauthorized")    row_id = str(uuid.uuid4())    path = unique_path(f"uploads/{user.id}/{args.file.name}")    await db.uploads.insert(id=row_id, owner=user.id, path=path, status="pending")    # state crosses in the completion token, so the callback can read it without a lookup.    return BeforeUploadResult(path=path, state={"row_id": row_id})async def finish_attachment(args):    await index_for_search(args.path, args.content_type)    await notify_owner(args.state["row_id"])    # Last. Everything above has to be done before the row stops looking abandoned.    await db.uploads.update(args.state["row_id"], status="ready", size=args.size, url=args.url)    return {"row_id": args.state["row_id"]}attachment = UploadRoute(    max_size="20mb",    content_types=["image/*", "application/pdf"],    on_before_upload=begin_attachment,    on_upload_complete=finish_attachment,)uploads = AsyncUploadHandler(routes={"attachment": attachment})

// uploads.gotype pending struct {	RowID string `json:"rowId"`}attachment := blob.UploadRoute[struct{}]{	ContentTypes: []string{"image/*", "application/pdf"},	MaxSize:      20 * blob.MB,	OnBeforeUpload: func(		a blob.BeforeUploadArgs[struct{}],	) (blob.BeforeUploadResult, error) {		user := getUser(a.Request)		if user == nil {			return blob.BeforeUploadResult{}, blob.NewError(				blob.CodeUnauthorized,				blob.ErrorOptions{},			)		}		rowID := uuid.NewString()		path := blob.UniquePath("uploads/" + user.ID + "/" + a.File.Name)		ctx := a.Request.Context()		err := db.InsertPendingUpload(ctx, rowID, user.ID, path)		if err != nil {			return blob.BeforeUploadResult{}, err		}		// State crosses in the completion token, so the callback can read		// it without a lookup.		return blob.BeforeUploadResult{			Path:  path,			State: pending{RowID: rowID},		}, nil	},	OnUploadComplete: func(a blob.UploadCompleteArgs[struct{}]) (any, error) {		var s pending		if err := json.Unmarshal(a.State, &s); err != nil {			return nil, err		}		ctx := a.Request.Context()		if err := indexForSearch(ctx, a.Path, a.ContentType); err != nil {			return nil, err		}		if err := notifyOwner(ctx, s.RowID); err != nil {			return nil, err		}		// Last. Everything above has to be done before the row stops		// looking abandoned.		if err := db.MarkUploadReady(ctx, s.RowID, a.Size, a.URL); err != nil {			return nil, err		}		return map[string]string{"rowId": s.RowID}, nil	},}uploads, err := blob.NewUploadHandler(blob.UploadHandlerOptions[struct{}]{	Routes: map[string]blob.UploadRoute[struct{}]{"attachment": attachment},})

Insert a pending row in the before-upload callback

It runs once per upload, before anything is signed, and already knows the path.

Flip it to ready as the last thing the completion callback does

Do all other work first. If the flip came earlier, a crash after it would leave a ready row whose work never happened.

Sweep rows still pending from the same cron, after the abort

An indexed query on your own table, not a bucket scan. Each row names the exact path to check.

Note

The upload id first reaches your code in the completion callback, so the example mints its own row id in the before-upload callback and passes it as state, which reaches the completion callback.

The cron#

Add the row sweep to the route above. The abort runs first, and both halves share one window.

// app/api/cron/sweep-uploads/route.tsimport { Bucket } from "@upstash/blob"import { db } from "@/lib/db"const bucket = Bucket.fromEnv()const olderThan = "1d"export const GET = async () => {  const aborted = await bucket.abortStaleMultipartUploads({ olderThan, prefix: "uploads/" })  const rows = await db.uploads.findPending({ olderThan, limit: 500 })  if (rows.length) {    // del does not throw for a path with nothing at it, so a PUT that never finished    // and parts the abort above just cleared are both covered by the same call.    await bucket.del(rows.map((r) => r.path))    await db.uploads.delete(rows.map((r) => r.id))  }  return Response.json({ aborted: aborted.length, swept: rows.length })}

# app/cron.pyfrom upstash_blob import Bucketfrom app.db import dbbucket = Bucket.from_env()OLDER_THAN = "1d"def sweep_uploads() -> dict:    aborted = bucket.abort_stale_multipart_uploads(older_than=OLDER_THAN, prefix="uploads/")    rows = db.uploads.find_pending(older_than=OLDER_THAN, limit=500)    if rows:        # delete does not raise for a path with nothing at it, so a PUT that never finished        # and parts the abort above just cleared are both covered by the same call.        bucket.delete([r.path for r in rows])        db.uploads.delete([r.id for r in rows])    return {"aborted": len(aborted), "swept": len(rows)}

// cron.goconst olderThan = 24 * time.Hourfunc sweepUploads(ctx context.Context) (aborted, swept int, err error) {	stale, err := bucket.AbortStaleMultipartUploads(ctx, olderThan, "uploads/")	if err != nil {		return 0, 0, err	}	rows, err := db.FindPendingUploads(ctx, olderThan, 500)	if err != nil {		return len(stale), 0, err	}	if len(rows) > 0 {		paths := make([]string, len(rows))		ids := make([]string, len(rows))		for i, r := range rows {			paths[i], ids[i] = r.Path, r.ID		}		// DeleteMany does not fail for a path with nothing at it, so a PUT		// that never finished and parts the abort above just cleared are		// both covered by the same call.		if err := bucket.DeleteMany(ctx, paths); err != nil {			return len(stale), 0, err		}		if err := db.DeleteUploads(ctx, ids); err != nil {			return len(stale), 0, err		}	}	return len(stale), len(rows), nil}

Two things this example depends on:

One path, one upload. The sweep deletes whatever stands at a pending row’s path, so a path must never be shared. Unique paths give you that. With a reused path, a retry that succeeded leaves a ready row and an older pending row pointing at the same object, and the sweep would delete the good file. If you must reuse paths, skip rows whose path also has a ready row.
Abort first, one window. A route without multipart always on still uses parts for a file over the threshold, so a pending row can belong to a paused multipart upload with nothing at its path yet. Running the abort first with the same window means the parts are gone before the row is, so a late resume fails instead of completing against a row that no longer exists. A row sweep on a shorter window would delete rows out from under uploads that can still resume.

Without a pending row#

const page = await bucket.list({ prefix: "uploads/", limit: 1000 })const known = new Set(await recordedPaths())const cutoff = Date.now() - 24 * 60 * 60 * 1000const orphans = page.blobs.filter((b) => !known.has(b.path) && b.uploadedAt.getTime() < cutoff)if (orphans.length) await bucket.del(orphans.map((b) => b.path))

from datetime import datetime, timedelta, timezonepage = bucket.list(prefix="uploads/", limit=1000)known = set(recorded_paths())cutoff = datetime.now(timezone.utc) - timedelta(days=1)orphans = [b.path for b in page.blobs if b.path not in known and b.uploaded_at < cutoff]if orphans:    bucket.delete(orphans)

opts := &blob.ListOptions{Prefix: "uploads/", Limit: 1000}page, err := bucket.List(ctx, opts)if err != nil {	return err}known := recordedPaths() // map[string]boolcutoff := time.Now().Add(-24 * time.Hour)var orphans []stringfor _, b := range page.Blobs {	if !known[b.Path] && b.UploadedAt.Before(cutoff) {		orphans = append(orphans, b.Path)	}}if len(orphans) > 0 {	err = bucket.DeleteMany(ctx, orphans)}

An app that will not take a database write on the upload path can list the prefix and diff it against the rows it does have. This is weaker: list carries no metadata, so it cannot tell which upload wrote an object, and it pages through every object under the prefix to find the few that do not belong. Use it only when a pending row is not an option.

Do not let the completion callback throw on a database error. Any throw deletes the object, and the browser’s retried end then reports a 404. See Retries and throws.
Use unique paths unless overwriting is the intent. Two single-PUT uploads to the same path race, and the loser’s end fails with not_found even though its bytes landed. A unique path is the fix; see Unique paths.
An explicit cancel() is already handled. It tells your route, which aborts the multipart upload or deletes the single-PUT object. The gap is everything that is not an explicit cancel: a crash, a closed tab, a lost network.

The simplest fix: always multipart#

Sweeping incomplete multipart uploads#

Single PUT: the pending row#

The cron#

Without a pending row#

Related traps#