Katta fayllarni stream bilan yuklash: busboy, backpressure va S3 multipart

Qisqacha: Katta faylni xotiraga ham, server diskiga ham yig'mang. Uni oqim (stream) sifatida qabul qilib, darhol omborga uzating. Kutubxona busboy multipart tanani bo'laklab beradi. Siz boshidagi 4 100 baytdan turni tekshirasiz, oqim o'tayotganda sha256 va hajmni hisoblaysiz, @aws-sdk/lib-storage esa faylni 5 MB lik qismlar bilan yuklaydi. Kanon o'lchovida 200 MB fayl Multer xotira rejimida server xotirasini +405 MB oshirdi, oqim bilan esa +42 MB va eng tez.

Bu darsda

• Uch yuklash usulini (xotira, disk, oqim) server xotirasi va vaqt o'lchovi bilan solishtirasiz.
• busboy hodisalari va limitlari bilan multipart'ni oqim sifatida o'qiysiz.
• Oqim boshidan turni tekshirib, oqimni yo'qotmasdan davom ettirasiz.
• Oqim o'tayotganda sha256 va hajmni hisoblaysiz, backpressure qayerda ishlashini ko'rasiz.
• Limit oshsa yoki mijoz uzilsa, yuklashni bekor qilib, omborda chala qism qoldirmaysiz.

Oldin bilishingiz kerak: Fayl yuklash xavfsizligi, pipeline, finished va stream xatolari, Backpressure, O'z Readable/Transform stream'ingizni yozish.

Oldingi darsda fayl turini, nomini va kalitini tekshiradigan xavfsizlik.ts modulini yozdik. Endi uni haqiqiy yuklash oqimiga ulaymiz — 200 MB lik video ham server xotirasini to'ldirmasin.

1. Nega bu kerak?

Malika vazifaga 200 MB lik videoni biriktirmoqchi — mijoz bilan uchrashuv yozuvi. Birinchi darsdagi Multer memoryStorage bilan nima bo'ladi? Fayl to'liq xotiraga yig'iladi. O'nta foydalanuvchi bir vaqtda shunday yuklasa — 2 GB.

diskStorage xotirani tejaydi, lekin fayl ikki marta yoziladi: avval server diskiga, keyin omborga. Bir nechta server bo'lsa, disklari alohida — fayl qaysi serverda qolgani noma'lum.

Uchinchi yo'l — oqim: baytlar kelishi bilan omborga uzatiladi, serverda faqat kichik bufer turadi. O'xshatish: suvni chelakda tashish (xotira), katta bakka to'ldirib keyin quvur bilan uzatish (disk) va to'g'ridan quvur ulash (oqim).

1.1 Kanon o'lchovi: 200 MB

Kanon uch usulni sinadi. Har usul alohida server jarayonida ishladi (boshqasining qoldig'i aralashmasin), server har 10 ms da o'z xotirasini (RSS) yozdi, mijoz esa 200 MB matn faylini diskdan oqim bilan yubordi. Har usul ikki marta. Har qatorda ikki joyga qarang: server … ms (vaqt) va oxirgi qavsdagi +… MB (xotira qancha oshdi):

memory  #1: 200 hajm 200.0 MB, server 3662 ms (mijoz 3694 ms), RSS 89.2 MB → eng yuqori 494.7 MB (+405.5 MB)
memory  #2: 200 hajm 200.0 MB, server 4826 ms (mijoz 4861 ms), RSS 89.6 MB → eng yuqori 495.6 MB (+406.0 MB)
disk    #1: 200 hajm 200.0 MB, server 4273 ms (mijoz 4283 ms), RSS 89.8 MB → eng yuqori 125.8 MB (+36.0 MB)
disk    #2: 200 hajm 200.0 MB, server 4120 ms (mijoz 4129 ms), RSS 89.5 MB → eng yuqori 125.1 MB (+35.6 MB)
oqim    #1: 200 hajm 200.0 MB, server 3392 ms (mijoz 3399 ms), RSS 90.6 MB → eng yuqori 132.2 MB (+41.6 MB), progress hodisalari 40
oqim    #2: 200 hajm 200.0 MB, server 3136 ms (mijoz 3149 ms), RSS 89.6 MB → eng yuqori 135.4 MB (+45.8 MB), progress hodisalari 40

Grafikka qarang: xotira rejimi fayldan ikki baravar ko'p xotira oldi — Multer bo'laklarni yig'ib, oxirida bitta katta Buffer ga birlashtiradi. Disk va oqim deyarli teng, lekin oqim tezroq: 3.1–3.4 s, disk esa 4.1–4.3 s (avval diskka, keyin diskdan omborga). Lokal o'lchov — haqiqiy tarmoqda farq boshqacha bo'ladi, lekin yo'nalishi shu.

Tekshirib ko'ring: Nega oqim rejimida xotira 0 emas, +42 MB?

2. busboy: multipart oqimi

2.1 Multer'siz

Multer ichida busboy ishlashini birinchi darsda ko'rgan edik. Endi uni to'g'ridan ishlatamiz: busboy har fayl uchun Readable oqim beradi, nima qilishni esa biz hal qilamiz.

npm i busboy@1.6.0

import busboy from "busboy";

const bb = busboy({
  headers: req.headers, // boundary shu yerdan olinadi
  limits: {
    fileSize: maksHajm, files: 1, fields: 10,
    fieldSize: 10_000, parts: 12,
  },
});

bb.on("field", (nom, qiymat) => { /* matn maydoni */ });
bb.on("file", (nom, fayl, info) => {
  // fayl — Readable; info: { filename, encoding, mimeType }
});
bb.on("close", () => { /* tana tugadi */ });
bb.on("error", (e) => { /* buzilgan multipart */ });
req.pipe(bb);

• busboy({ headers }) — Content-Type dan boundary'ni oladi. Boundary bo'lmasa, shu qatorning o'zi xato tashlaydi (Multipart: Boundary not found) — shuning uchun kanon uni try ichida chaqiradi.
• req.pipe(bb) — so'rov tanasini busboy'ga uzatish. req ham oqim (Stream nima).
• "file" hodisasi — fayl qismi boshlandi. Uchinchi argument info da filename va mimeType — mijoz matni, oldingi darsdagidek.

2.2 Limitlar va ularning hodisalari

busboy limitlari Multer'nikiga o'xshaydi (Multer ularni busboy'ga uzatadi), lekin xato o'rniga hodisa chiqadi:

fileSize oshganda busboy faylni jim kesib qo'yadi: oqim tugaydi, lekin to'liq emas. "limit" ni tinglamasangiz, chala faylni "muvaffaqiyatli" saqlaysiz. Kanon modul har hodisani xato bilan yakunlaydi.

2.3 Oqimni oxirigacha o'qish majburiy

Kutilmagan maydondagi faylni rad etdingiz — endi uni tashlab ketsa bo'ladimi? Yo'q. busboy keyingi qismga o'tishi uchun shu faylning baytlari o'qib bo'linishi kerak. Hech kim o'qimasa, oqim to'xtaydi (Backpressure) va so'rov osilib qoladi. Shuning uchun kanon har rad etishda fayl.resume() chaqiradi — "baytlarni o'qib, tashlab yubor".

3. Turni oqim boshida tekshirish

3.1 Muammo

turniAniqla faylning birinchi 4 100 baytini so'raydi. Oqimda esa fayl bo'laklab keladi va bo'lak bir marta o'qilsa, oqimda qolmaydi. Boshini o'qib tekshirdik — endi omborga nima yuboramiz? Bosh qismi yo'qolmasligi kerak.

Yechim — uch qadam:

1. Bo'laklarni yig'amiz, toki jami ≥ 4 100 bayt bo'lguncha. Keyin oqimni to'xtatamiz (pause).
2. Yig'ilgan boshni tekshiramiz.
3. Yangi oqimga avval boshni yozamiz, keyin asl oqimning qolganini ulaymiz.

Bu g'oyani async iteratorlar bilan brauzerda ham ko'rish mumkin. Fayl — 8 baytli PNG imzosi va beshta 2 000 baytli bo'lak:

async function* faylOqimi() {
  yield new Uint8Array([137, 80, 78, 71, 13, 10, 26, 10]);
  for (let i = 0; i < 5; i++) yield new Uint8Array(2000);
}

async function boshiniOl(iter, n) {
  const bolaklar = [];
  let uzunlik = 0;
  while (uzunlik < n) {
    const { value, done } = await iter.next();
    if (done) break;
    bolaklar.push(value);
    uzunlik += value.length;
  }
  const bosh = new Uint8Array(uzunlik);
  let joy = 0;
  for (const b of bolaklar) { bosh.set(b, joy); joy += b.length; }
  return bosh;
}

async function* qaytaUla(bosh, iter) {
  yield bosh; // avval o'qilgan bosh
  yield* iter; // keyin qolgani
}

const iter = faylOqimi();
const bosh = await boshiniOl(iter, 4100);
const imzo = bosh.slice(0, 4).join();
console.log("bosh:", bosh.length, "bayt; imzo:", imzo);

let jami = 0;
let soni = 0;
for await (const b of qaytaUla(bosh, iter)) {
  jami += b.length;
  soni++;
}
console.log("omborga:", jami, "bayt,", soni, "bo'lakda");

Konsolda:

bosh: 6008 bayt; imzo: 137,80,78,71
omborga: 10008 bayt, 3 bo'lakda

Ikki narsaga e'tibor bering:

• Bosh 4 100 dan katta chiqdi (6 008): bo'laklarni kesmaymiz, chegarani o'tgan bo'lak to'liq olinadi. Bu xato emas — tekshiruvga ko'proq bayt berildi.
• Omborga hamma bayt yetdi: 8 + 5 × 2 000 = 10 008. Bosh — bitta bo'lak, qolgan ikki bo'lak asl oqimdan.

3.2 Node'da: pause va PassThrough

Kanon modulda xuddi shu g'oya Node oqimlari bilan (parcha, xato xabari qisqartirilgan):

const { bosh, tugadi: qisqa } = await boshiniOl(fayl, BOSH_BAYT);
const tur = await turniAniqla(bosh, info.mimeType);
if (!tur.ok) {
  fayl.resume();
  throw new YuklashXato(415, tur.sabab, "Tur qabul qilinmaydi");
}
const sanagich = new Sanagich();
const kirish = new PassThrough();
kirish.pipe(sanagich);
kirish.write(bosh);       // 1) yig'ilgan bosh
if (qisqa) kirish.end();  // fayl 4 100 baytdan kichik edi
else fayl.pipe(kirish);   // 2) qolgani — asl oqimdan

• boshiniOl (Node versiyasi) "data" hodisasini tinglaydi, yetarli yig'ilgach oqim.pause() qiladi va tinglovchilarni olib tashlaydi. Fayl 4 100 baytdan kichik bo'lsa — tugadi: true.
• PassThrough — kirganini o'zgartirmasdan chiqaradigan oqim. Unga avval bosh yoziladi, keyin fayl.pipe(kirish) qolganini ulaydi — pipe to'xtatilgan oqimni o'zi qayta yurgizadi.
• Tur rad etilsa — fayl.resume() va 415 (Unsupported Media Type).

Tekshirib ko'ring: Nega turni tekshirish uchun butun faylni kutmaymiz?

4. Oqim o'tayotganda hisoblash: sha256 va hajm

Fayl omborga ketyapti — uning hajmini va sha256 hash'ini (crypto: hash) qanday bilamiz? Faylni qayta o'qimasdan, oqim o'tayotganda. Buning uchun o'z Transform oqimimiz (O'z Transform stream'ingiz): har bo'lakni hash'ga qo'shadi, sanaydi va o'zgartirmasdan uzatadi.

import { Transform, Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";
import { createHash } from "node:crypto";

class Sanagich extends Transform {
  hash = createHash("sha256");
  bayt = 0;
  _transform(bolak, _enc, cb) {
    this.hash.update(bolak);
    this.bayt += bolak.length;
    cb(null, bolak); // bo'lak o'zgarmasdan keyingisiga
  }
}

const sanagich = new Sanagich();
const ombor = new Writable({
  write(_bolak, _enc, cb) { cb(); }, // soxta ombor
});
await pipeline(
  Readable.from([Buffer.from("salom "), Buffer.from("dunyo")]),
  sanagich,
  ombor,
);

const butun = createHash("sha256")
  .update("salom dunyo")
  .digest("hex");
const oqimdan = sanagich.hash.digest("hex");
console.log(sanagich.bayt, "bayt");
console.log(oqimdan.slice(0, 16));
console.log(oqimdan === butun);

Konsolda:

11 bayt
3d241073bae411a4
true

Bo'laklab hisoblangan hash butun matnnikiga teng. Bu hash nima uchun kerak? Bazada bir foydalanuvchining bir xil fayllarini aniqlash (dublikat) uchun — Fayl metama'lumoti darsida. Va yuklangan faylning butunligini keyin tekshirish uchun.

5. Omborga oqim: @aws-sdk/lib-storage

5.1 Hajmi noma'lum oqim

Omborni (S3 yoki uning lokal o'xshashi MinIO) keyingi darsda to'liq o'rganamiz. Hozir bitta narsani bilish yetarli: kanonning FaylOmbori klassida yoz(kalit, oqim, tur, sozlama) metodi bor, u oqimni omborga yozadi.

Bir muammo bor. S3 ga oddiy yuklashda (PutObject) fayl hajmini oldindan aytish kerak (Content-Length). Oqimda esa hajm oxirida ma'lum bo'ladi. Yechim — multipart upload (bo'laklab yuklash): fayl 5 MB lik qismlarga bo'linadi, har biri alohida so'rov bilan yuboriladi, oxirida ombor ularni bitta faylga yig'adi. Brauzer tomonidagi "bo'laklab yuklash" g'oyasini Fayl yuklash va progress darsida ko'rgansiz — bu uning server tomoni.

@aws-sdk/lib-storage ning Upload klassi buni o'zi qiladi:

npm i @aws-sdk/client-s3@3.1153.0 @aws-sdk/lib-storage@3.1153.0

const up = new Upload({
  client: this.s3,
  params: { Bucket, Key: kalit, Body: oqim, ContentType: tur },
  partSize: 5 * 1024 * 1024, // har qism 5 MB
  queueSize: 2,              // bir vaqtda 2 ta qism
  leavePartsOnError: false,  // xatoda chala qismlarni o'chir
  abortController,           // tashqaridan bekor qilish
});
up.on("httpUploadProgress", (h) => progress(h.loaded ?? 0));
await up.done();

• partSize — 5 MB S3 ning minimal qism hajmi (oxirgi qismdan tashqari). Kanon 1 MB qo'yib ko'rdi — EntityTooSmall xatosi (keyingi darsda).
• queueSize: 2 — bir vaqtda ikki qism yuboriladi. Xotira ≈ partSize × queueSize = 10 MB, fayl 2 GB bo'lsa ham.
• httpUploadProgress — har qism yuklanganda. 200 MB / 5 MB = 40 qism → kanon o'lchovida 40 progress hodisasi.

5.2 Backpressure bu yerda qanday ishlaydi

flowchart LR
  M[Mijoz] -->|TCP| R[req]
  R --> B[busboy]
  B --> P[PassThrough]
  P --> S[Sanagich: sha256]
  S --> U[Upload: 2 x 5 MB]
  U --> O[(Ombor)]
  U -.->|to'la: kut| S
  S -.-> P
  P -.-> B
  B -.-> R
  R -.->|sekinroq| M

Diagrammaga qarang: to'liq chiziq — baytlar yo'li, uzuq chiziq — "to'xta" signali. Ombor sekinlashsa, signal zanjir bo'ylab orqaga, mijozgacha boradi.

Upload ikkita qismni yuborib turgan paytda yangi bo'lak olmaydi. PassThrough va Sanagich buferlari to'ladi, pipe busboy oqimini to'xtatadi, busboy esa req ni. Oxirida TCP mijozga "sekinroq" deydi. Bu zanjir Backpressure darsidagi write() → false → drain mexanizmining o'zi.

Kanon buni alohida o'lchadi: tez o'quvchi (20 MB fayl) va sekin yozuvchi (har bo'lak 5 ms), highWaterMark 64 KB:

pipeline: 20 MB, 4992 ms, yozuvchi buferi eng ko'pi 0.1 MB, o'quvchi pause() 320 marta
write() natijasiga qaramasdan: yozuvchi buferida 19.6 MB navbatda qoldi (eng ko'pi 19.6 MB) — backpressure e'tiborsiz

pipeline bilan bufer 0.1 MB dan oshmadi — o'quvchi 320 marta to'xtatildi. write() qaytargan false ga qaramagan kod esa butun faylni (19.6 MB) xotirada navbatga yig'di. Katta fayl yuklashda bu farq — server yashaydimi yoki yo'qmi.

Qoida: oqimlarni pipe/pipeline bilan ulang, "data" hodisasida qo'lda write() qilmang.

6. Bekor qilish: chala qism qolmasin

6.1 Ikki holat

Yuklash yarmida to'xtashi mumkin:

1. Limit oshdi. 1 MB limitli marshrutga 6 MB keldi — busboy "limit" hodisasini chiqaradi.
2. Mijoz uzildi. Foydalanuvchi sahifani yopdi, internet uzildi. req ning "close" hodisasi keladi, lekin req.complete — false.

Ikkala holatda ham omborda allaqachon bir nechta 5 MB lik qism yuklangan bo'lishi mumkin. Multipart upload tugallanmasa, bu qismlar ko'rinmaydi, lekin joy egallaydi (va AWS'da pul to'lanadi).

6.2 AbortController bilan

Kanon modulda bitta AbortController (pipeline va bekor qilish) va bitta bitir funksiyasi bor (parcha, xabarlar qisqartirilgan):

const ac = new AbortController();
let tugadi = false;
const bitir = (x: YuklashXato) => {
  if (!tugadi) { tugadi = true; ac.abort(x); yoq(x); }
};

fayl.on("limit", () => {
  bitir(new YuklashXato(413, "katta-fayl", "Fayl katta"));
  fayl.resume();
});
req.on("close", () => {
  if (!req.complete) {
    bitir(new YuklashXato(499, "mijoz-uzildi", "Uzildi"));
  }
});

• ac.abort(x) → Upload bekor bo'ladi → leavePartsOnError: false tufayli SDK AbortMultipartUpload yuboradi → yuklangan qismlar o'chadi.
• tugadi bayrog'i — Promise faqat bir marta rad etilsin: limit va uzilish birga kelishi mumkin.
• 499 — HTTP standartida yo'q. Uni nginx "mijoz so'rovni yopdi" uchun ishlatadi; kanon ham shu ma'noda oldi. Javob baribir mijozga yetmaydi — kod log uchun.

Kanon testi (MinIO bilan) 6 MB faylni 1 MB limitga yubordi va keyin ikki narsani tekshirdi: obyektlar soni o'zgarmadi va ListMultipartUploads — 0 chala yuklash.

Diqqat: MinIO AbortIncompleteMultipartUpload lifecycle qoidasini rad etdi (kanon topilmasi). AWS S3 da esa bucket'ga "N kundan eski chala yuklashlarni o'chir" qoidasi qo'yiladi — kod xato qilsa ham zaxira. Ikkalasiga ham tayaning: kod bekor qiladi, qoida qolganini tozalaydi.

6.3 Xatolar xaritasi

7. Resumable yuklash: tus (tanishuv)

200 MB yuklashning 90 % ida internet uzilsa — hammasi boshidan. Mobil internetda bu og'riqli. Resumable upload (davom ettiriladigan yuklash) — server qaysi baytgacha olganini eslab qoladi, mijoz uzilishdan keyin o'sha joydan davom etadi.

Eng mashhur ochiq protokol — tus: mijoz POST bilan yuklashni yaratadi, keyin PATCH bilan bo'laklarni Upload-Offset sarlavhasi bilan yuboradi, uzilsa HEAD bilan "qayergacha oldingiz?" deb so'raydi. Node uchun @tus/server, brauzer uchun tus-js-client va Uppy bor.

Ikkinchi yo'l — S3 multipart'ni brauzerning o'zi qilishi: har qism uchun alohida presigned URL (Presigned URL'lar darsida kanon buni sinaydi).

Diqqat: tus kanonda sinalmagan — tavsif protokol hujjati bo'yicha (tus.io, 2026-10-28 da o'qildi). vazifalar uchun 10 MB lik ilovalarga resumable kerak emas; video platformasi qursangiz — kerak.

8. Hujumchi nigohi

Oqimli yuklash — uzoq ochiq turadigan ulanish. Hujumchi buni ishlatadi:

• Sekin yuklash (slowloris). Sekundiga 1 bayt yuboradi — ulanish soatlab band. Himoya — server timeoutlari (Server timeoutlari) va reverse proxy limitlari.
• Cheksiz tana. fileSize limiti busboy'da — oshgan zahoti 413 va oqim to'xtaydi.
• Ko'p mayda qism. 10 000 ta bo'sh maydon — parts: 12, fields: 10.
• Ombor to'ldirish. Yuklashni boshlab, yarmida uzish — chala qismlar qoladi. Himoya — leavePartsOnError: false va lifecycle qoidasi.
• Turini yashirish. exe — busboy mimeType: "image/png" deydi, lekin turniAniqla boshidan tekshiradi va 415.

Biz kanon modulini soxta ombor bilan ishga tushirdik (MinIO'siz; ombor baytlarni sanab tashlaydi, Node 24.21.0). Natija:

  ombor.yoz: vaqtincha/7/<uuid>.png, image/png, 5000 bayt
png: 201 {"kalit":"<kalit>","aslNom":"rasm (1).png","mime":"image/png","hajm":5000,"sha256":"7aad6d3f66abe45236dc1383277b25665a628159a5b56a8f4fa73e88204902d9","maydonlar":{"izoh":"salom"}}
  ombor.yoz: vaqtincha/7/<uuid>.txt, text/plain, 65000 bayt
matn: 201 {"kalit":"<kalit>","aslNom":"uzun.txt","mime":"text/plain","hajm":65000,"sha256":"ca908e3349b4dcb3596c410ed590300323d2040eecf7fae94acb8fa394301c17","maydonlar":{"izoh":"salom"}}
exe: 415 {"kod":"ruxsat-yoq"}
katta: 413 {"kod":"katta-fayl"}
boshqa maydon: 400 {"kod":"kutilmagan-maydon"}
json: 400 {"kod":"multipart-emas"}

PNG nomi ../../rasm (1).png edi — aslNom da faqat rasm (1).png qoldi, kalit esa tasodifiy. Mijoz application/octet-stream dedi, ombor image/png oldi — tur baytlardan. 65 000 baytli matn 4 100 lik boshdan ancha katta — hammasi yetdi. exe, katta fayl va JSON so'rov omborga umuman yetmadi.

9. Ko'p uchraydigan xatolar

10. Mashqlar

Mashqlar kurs/mashqlar/29/03-oqim/ da.

1-mashq (oson): bo'shliqlarni to'ldiring

• S3 multipart'ning minimal qism hajmi — MB.
• partSize 5 MB va queueSize 2 bo'lsa, yuklash buferi taxminan MB.
• Rad etilgan fayl oqimini o'qib tashlash uchun fayl.().
• Fayl limitdan oshsa busboy fayl oqimida hodisasini chiqaradi.
• Xatoda chala qismlarni o'chirish sozlamasi — leavePartsOnError:

2-mashq (o'rta): limitli sanagich

Darsdagi Sanagich ni LimitliSanagich ga aylantiring: konstruktor maks baytni oladi; jami bayt maks dan oshsa, cb(new RangeError(...)) bilan to'xtasin (xabar: "limit: <maks> bayt"). pipeline bilan ikki marta sinang: maks = 20 va "salom " + "dunyo" (11 bayt), keyin maks = 8. Har sinovda yo "OK <bayt>", yo xato xabarini chiqaring. Ishora: pipeline xatoni await da throw qiladi (pipeline va stream xatolari).

import { Transform, Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";

class LimitliSanagich extends Transform {
  bayt = 0;
  constructor(maks) {
    super();
    this.maks = maks;
  }
  _transform(bolak, _enc, cb) {
    this.bayt += bolak.length;
    if (this.bayt > this.maks) {
      cb(new RangeError(`limit: ${this.maks} bayt`));
      return;
    }
    cb(null, bolak);
  }
}

async function sina(maks) {
  const s = new LimitliSanagich(maks);
  try {
    await pipeline(
      Readable.from([Buffer.from("salom "), Buffer.from("dunyo")]),
      s,
      new Writable({ write(_b, _e, cb) { cb(); } }),
    );
    console.log("OK", s.bayt);
  } catch (e) {
    console.log(e.name, e.message);
  }
}

await sina(20);
await sina(8);

OK 11
RangeError limit: 8 bayt

3-mashq (qiyin): progress foizi

Kanon Upload har qismdan keyin yuklandi (bayt) ni beradi. progressYozuvchi(jami) funksiyasini yozing: u (yuklandi) => … funksiyasini qaytarsin. Har chaqiruvda foiz butun songa yaxlitlanib chiqsin, lekin faqat foiz o'zgarganda. jami noma'lum (undefined) bo'lsa — foiz o'rniga MB ni bir xona aniqlikda chiqarsin. 12 MB faylni 5 MB lik qismlar bilan taqlid qiling (yuklandi: 5, 10, 12 MB) va jami noma'lum holatni ham (2.5 MB, 5 MB) sinang. Ishora: yopilish (Closure) oxirgi foizni eslab qoladi.

function progressYozuvchi(jami) {
  let oxirgi = -1;
  return (yuklandi) => {
    if (jami === undefined) {
      console.log(`${(yuklandi / 1024 / 1024).toFixed(1)} MB`);
      return;
    }
    const foiz = Math.round((yuklandi / jami) * 100);
    if (foiz === oxirgi) return;
    oxirgi = foiz;
    console.log(`${foiz}%`);
  };
}

const MB = 1024 * 1024;
const p = progressYozuvchi(12 * MB);
for (const y of [5 * MB, 10 * MB, 10 * MB, 12 * MB]) p(y);

const noma = progressYozuvchi(undefined);
noma(2.5 * MB);
noma(5 * MB);

42%
83%
100%
2.5 MB
5.0 MB

11. Loyiha qadami: yuklash.ts moduli

Kanon modul — modullar/src/fayl/yuklash.ts. U oldingi darsdagi xavfsizlik.ts ni va keyingi darsdagi FaylOmbori ni ishlatadi. Telefon uchun qatorlarni bo'ldik; ma'no kanondagidek (faqat tozalangan nom bir marta hisoblanib, aslNom o'zgaruvchisiga olindi):

// yuklash.ts — multipart faylni xotiraga to'plamasdan, busboy →
// (tur tekshiruvi + sha256 + hajm) → S3 multipart oqimi.
// Xotira: fayl hajmidan qat'iy nazar ~ partSize × queueSize
// (5 MB × 2). Limit oshsa yoki mijoz uzilsa — S3 yuklash bekor
// qilinadi (`leavePartsOnError: false` → AbortMultipartUpload),
// bucket'da chala qism qolmaydi.
import type { IncomingMessage } from "node:http";
import {
  PassThrough, Transform, type Readable, type TransformCallback,
} from "node:stream";
import { createHash, type Hash } from "node:crypto";
import busboy from "busboy";
import {
  BOSH_BAYT, faylNomiTozala, turniAniqla,
} from "./xavfsizlik.ts";
import type { FaylOmbori } from "./s3.ts";

export class YuklashXato extends Error {
  readonly status: 400 | 413 | 415 | 499;
  readonly kod: string;
  constructor(
    status: 400 | 413 | 415 | 499, kod: string, xabar: string,
  ) {
    super(xabar);
    this.name = "YuklashXato";
    this.status = status;
    this.kod = kod;
  }
}

export type YuklashNatija = {
  kalit: string;
  aslNom: string;
  mime: string;
  hajm: number;
  sha256: string;
  maydonlar: Record<string, string>;
};

export type YuklashSozlama = {
  ombor: FaylOmbori;
  maksHajm: number;
  /** kengaytma → saqlash kaliti
   * (masalan `saqlashKaliti(egasiId, k, "vaqtincha")`) */
  kalit: (kengaytma: string) => string;
  maydonNomi?: string;
  progress?: (yuklandi: number) => void;
};

/** sha256 va bayt sonini oqim o'tayotganda hisoblaydi. */
class Sanagich extends Transform {
  readonly hash: Hash = createHash("sha256");
  bayt = 0;
  override _transform(
    bolak: Buffer, _enc: BufferEncoding, cb: TransformCallback,
  ): void {
    this.hash.update(bolak);
    this.bayt += bolak.length;
    cb(null, bolak);
  }
}

/** Oqim boshidan kamida `n` bayt (yoki butun fayl, agar qisqa
 * bo'lsa) yig'adi va oqimni to'xtatib turadi. */
function boshiniOl(
  oqim: Readable, n: number,
): Promise<{ bosh: Buffer; tugadi: boolean }> {
  return new Promise((ok, yoq) => {
    const bolaklar: Buffer[] = [];
    let uzunlik = 0;
    const tozala = () => {
      oqim.off("data", onData);
      oqim.off("end", onEnd);
      oqim.off("error", onError);
    };
    const onData = (b: Buffer) => {
      bolaklar.push(b);
      uzunlik += b.length;
      if (uzunlik >= n) {
        oqim.pause();
        tozala();
        ok({ bosh: Buffer.concat(bolaklar), tugadi: false });
      }
    };
    const onEnd = () => {
      tozala();
      ok({ bosh: Buffer.concat(bolaklar), tugadi: true });
    };
    const onError = (e: Error) => { tozala(); yoq(e); };
    oqim.on("data", onData).on("end", onEnd).on("error", onError);
  });
}

export function oqimliYuklash(
  req: IncomingMessage, s: YuklashSozlama,
): Promise<YuklashNatija> {
  const maydonNomi = s.maydonNomi ?? "fayl";
  return new Promise<YuklashNatija>((ok, yoq) => {
    let bb: busboy.Busboy;
    try {
      bb = busboy({
        headers: req.headers,
        limits: {
          fileSize: s.maksHajm, files: 1, fields: 10,
          fieldSize: 10_000, parts: 12,
        },
      });
    } catch {
      yoq(new YuklashXato(400, "multipart-emas",
        "Content-Type multipart/form-data bo'lishi kerak"));
      return;
    }
    const ac = new AbortController();
    const maydonlar: Record<string, string> = {};
    let ish: Promise<YuklashNatija> | undefined;
    let tugadi = false;
    const bitir = (x: YuklashXato) => {
      if (!tugadi) { tugadi = true; ac.abort(x); yoq(x); }
    };

    bb.on("field", (nom, qiymat) => { maydonlar[nom] = qiymat; });
    bb.on("filesLimit", () => {
      bitir(new YuklashXato(400, "kop-fayl",
        "Faqat bitta fayl yuboring"));
    });
    bb.on("fieldsLimit", () => {
      bitir(new YuklashXato(400, "kop-maydon",
        "Maydonlar juda ko'p"));
    });
    bb.on("partsLimit", () => {
      bitir(new YuklashXato(400, "kop-qism", "Qismlar juda ko'p"));
    });
    bb.on("file", (nom, fayl, info) => {
      if (nom !== maydonNomi || ish) {
        fayl.resume();
        bitir(new YuklashXato(400, "kutilmagan-maydon",
          `Fayl '${maydonNomi}' maydonida bo'lishi kerak`));
        return;
      }
      fayl.on("limit", () => {
        bitir(new YuklashXato(413, "katta-fayl",
          `Fayl ${String(s.maksHajm)} baytdan katta`));
        fayl.resume();
      });
      ish = (async () => {
        const { bosh, tugadi: qisqa } =
          await boshiniOl(fayl, BOSH_BAYT);
        const tur = await turniAniqla(bosh, info.mimeType);
        if (!tur.ok) {
          // qolganini o'qib tashlaymiz — aks holda so'rov
          // osilib qoladi
          fayl.resume();
          throw new YuklashXato(415, tur.sabab,
            "Fayl turi qabul qilinmaydi " +
              `(${tur.topildi ?? tur.sabab})`,
          );
        }
        const sanagich = new Sanagich();
        const kirish = new PassThrough();
        kirish.pipe(sanagich);
        kirish.write(bosh);
        if (qisqa) kirish.end();
        else fayl.pipe(kirish);
        const kalit = s.kalit(tur.kengaytma);
        const aslNom = faylNomiTozala(info.filename ?? "fayl");
        await s.ombor.yoz(kalit, sanagich, tur.mime, {
          signal: ac.signal,
          meta: { "asl-nom": encodeURIComponent(aslNom) },
          ...(s.progress ? { progress: s.progress } : {}),
        });
        return {
          kalit, aslNom, mime: tur.mime, hajm: sanagich.bayt,
          sha256: sanagich.hash.digest("hex"), maydonlar,
        };
      })();
      ish.catch((e: unknown) => {
        bitir(e instanceof YuklashXato ? e
          : new YuklashXato(400, "yuklash-xatosi",
            e instanceof Error ? e.message : String(e)));
      });
    });
    bb.on("close", () => {
      if (tugadi) return;
      if (!ish) {
        bitir(new YuklashXato(400, "fayl-yoq",
          `'${maydonNomi}' maydonida fayl yo'q`));
        return;
      }
      ish.then(
        (n) => {
          if (!tugadi) { tugadi = true; ok({ ...n, maydonlar }); }
        },
        () => { /* bitir() allaqachon chaqirilgan */ },
      );
    });
    bb.on("error", (e: unknown) => {
      bitir(new YuklashXato(400, "buzilgan-multipart",
        e instanceof Error ? e.message : String(e)));
    });
    req.on("close", () => {
      if (!req.complete) {
        bitir(new YuklashXato(499, "mijoz-uzildi",
          "Mijoz so'rovni oxirigacha yubormadi"));
      }
    });
    req.pipe(bb);
  });
}

Modul qanday ishlatiladi (kanon testidagi Express marshruti):

app.post("/fayl", (req, res) => {
  oqimliYuklash(req, {
    ombor,
    maksHajm: 1024 * 1024,
    kalit: (k) => saqlashKaliti(7, k, "vaqtincha"),
  }).then(
    (n) => { res.status(201).json(n); },
    (e: unknown) => {
      const x = e as YuklashXato;
      res.status(x.status).json({ kod: x.kod });
    },
  );
});

Marshrutda express.json() yoki Multer yo'q — tanani faqat oqimliYuklash o'qiydi. Kalit vaqtincha/ prefiksi bilan: fayl bazaga yozilib tasdiqlangandan keyin doimiy joyga ko'chiriladi (Fayl metama'lumoti).

Testlar — test/fayl-s3.test.ts, MinIO bilan. Ular tekshiradi:

• PNG → 201, sha256 to'g'ri;
• 4 100 baytdan katta matn to'liq yetadi;
• exe → 415, omborda hech narsa yo'q;
• 6 MB → 413, chala multipart 0;
• noto'g'ri so'rovlar → 400.

Kanonda hammasi o'tdi (modullar paketi 82/82). Bizda Docker yoqilmagan — MinIO'siz ishga tushirsak, S3 testlari o'tkazib yuboriladi, faqat ombor kerak bo'lmagan test ishlaydi:

✔ contentDisposition: UTF-8 nom va ASCII zaxira (1.3313ms)
﹣ S3 (MinIO) bilan (0.1515ms) # S3_ENDPOINT yo'q
ℹ tests 1
ℹ suites 1
ℹ pass 1
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 417.11

E'tibor bering: skipped 0, garchi butun describe o'tkazilgan bo'lsa ham. node:test o'tkazilgan to'plam ichidagi testlarni sanamaydi — "hammasi o'tdi" degan xulosani skipped ga qarab qilmang, ﹣ belgisini ko'ring. Shuning uchun «Hujumchi nigohi» bo'limida modulni soxta ombor bilan alohida yurgizdik.

Commit xabari (modul uchun):

git commit -m "feat(fayl): oqimliYuklash — busboy oqimi, tur, sha256, S3"

vazifalar-api ga marshrut sifatida ulash 28-qism (auth) tugagach bo'ladi.

12. Real ishda

• Video va katta hujjat platformalari (YouTube, Google Drive, Telegram) — faylni bo'laklab va davom ettiriladigan qilib qabul qiladi.
• Serverless (AWS Lambda) da tana hajmi cheklangan va oqim qiyin — u yerda brauzer to'g'ridan omborga yuklaydi (Presigned URL'lar).
• NestJS da ham katta fayl uchun Multer'dan voz kechib, busboy yoki presigned yo'l tanlanadi.
• Intervyuda: "2 GB faylni qanday qabul qilasiz?", "Backpressure nima va u yuklashda qayerda ishlaydi?", "Yarim yuklangan fayllarni qanday tozalaysiz?"

Xulosa

• Kanon o'lchovi (200 MB): Multer xotira +405 MB, disk +36 MB, busboy oqim → S3 +42 MB va eng tez.
• busboy — multipart oqimi; limitlar hodisa sifatida keladi, rad etilgan fayl resume() bilan o'qib tashlanadi.
• Tur oqim boshida (4 100 bayt) tekshiriladi, bosh PassThrough ga qaytarib yoziladi — bayt yo'qolmaydi.
• Transform oqim o'tayotganda sha256 va hajmni hisoblaydi.
• lib-storage Upload — 5 MB qismlar, xotira ≈ partSize × queueSize; pipe backpressure'ni TCP gacha uzatadi.
• Limit yoki uzilishda AbortController → chala multipart o'chadi; tus — davom ettiriladigan yuklash (tanishuv).

Keyingi dars: Object storage: S3 va MinIO — bucket, kalit, PutObject/GetObject, ETag, private va public bucket hamda kanonning FaylOmbori klassi.

Manbalar

• busboy (limitlar, limit hodisasi): https://github.com/mscdex/busboy
• AWS SDK v3 @aws-sdk/lib-storage (Upload, partSize, queueSize, leavePartsOnError): https://github.com/aws/aws-sdk-js-v3/tree/main/lib/lib-storage
• Amazon S3 multipart upload limitlari (minimal qism 5 MiB, 10 000 qism): https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html
• Node.js Streams — backpressure: https://nodejs.org/en/learn/modules/backpressuring-in-streams
• tus protokoli 1.0.0: https://tus.io/protocols/resumable-upload — sinalmagan, 2026-10-28 da o'qildi
• Kanon tajribasi (IlmHamroh, 2026-10-27): busboy 1.6.0, multer 2.4.0, @aws-sdk 3.1153.0, MinIO RELEASE.2025-09-07 (Docker, lokal), Node 24.21.0, i5-12500H. Soxta ombor bilan yurgizish va testlar — biz ishga tushirdik (2026-10-28).