Add video download throttling

config/default.yaml

@@ -20,6 +20,7 @@ secrets:
 http_timeouts:
   request: '5 minutes'
 
+# Limit number of HTTP requests per IP
 rates_limit:
   api:
     # 50 attempts in 10 seconds
@@ -555,6 +556,15 @@ nsfw_flags_settings:
   # using NSFW flags (violent content, etc.) set by video authors
   enabled: true
 
+download:
+  # Max cumulative download speed in bytes/s for all download requests, set to null for unlimited
+  # Supports byte format ("500KB", etc.)
+  max_total_bytes_per_second: 5MB
+
+  # Max cumulative download speed in bytes/s for a specific IP, set to null for unlimited
+  # Supports byte format ("500KB", etc.)
+  max_bytes_per_ip_per_second: 2MB
+
 download_generate_video:
   # Max parallel downloads on your instance
   # Each download spawns an ffmpeg process

config/dev.yaml

@@ -148,3 +148,6 @@ transcoding:
 user:
   password_constraints:
     min_length: 6
+
+download_generate_video:
+  max_parallel_downloads: 2

config/production.yaml.example

@@ -18,6 +18,7 @@ secrets:
 http_timeouts:
   request: '5 minutes'
 
+# Limit number of HTTP requests per IP
 rates_limit:
   api:
     # 50 attempts in 10 seconds
@@ -553,6 +554,15 @@ nsfw_flags_settings:
   # using NSFW flags (violent content, etc.) set by video authors
   enabled: true
 
+download:
+  # Max cumulative download speed in bytes/s for all download requests, set to null for unlimited
+  # Supports byte format ("500KB", etc.)
+  max_total_bytes_per_second: 5MB
+
+  # Max cumulative download speed in bytes/s for a specific IP, set to null for unlimited
+  # Supports byte format ("500KB", etc.)
+  max_bytes_per_ip_per_second: 2MB
+
 download_generate_video:
   # Max parallel downloads on your instance
   # Each download spawns an ffmpeg process

packages/tests/src/api/server/config.ts

@@ -146,6 +146,9 @@ function checkInitialConfig (server: PeerTubeServer, data: CustomConfig) {
 
   expect(data.storyboards.enabled).to.be.true
 
+  expect(data.download.maxTotalBytesPerSecond).to.equal(2 * 1024 * 1024)
+  expect(data.download.maxBytesPerIpPerSecond).to.be.null
+
   expect(data.export.users.enabled).to.be.true
   expect(data.export.users.exportExpiration).to.equal(1000 * 3600 * 48)
   expect(data.export.users.maxUserVideoQuota).to.equal(10737418240)
@@ -440,6 +443,10 @@ function buildNewCustomConfig (server: PeerTubeServer): CustomConfig {
         enabled: true
       }
     },
+    download: {
+      maxTotalBytesPerSecond: 3 * 1024 * 1024,
+      maxBytesPerIpPerSecond: 512 * 1024
+    },
     export: {
       users: {
         enabled: false,

packages/tests/src/api/videos/video-download-throttle.ts

@@ -0,0 +1,153 @@
+/* oxlint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import { getHLS } from '@peertube/peertube-core-utils'
+import { HttpStatusCode } from '@peertube/peertube-models'
+import {
+  PeerTubeServer,
+  cleanupTests,
+  createSingleServer,
+  makeRawRequest,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  waitJobs
+} from '@peertube/peertube-server-commands'
+import { expect } from 'chai'
+
+describe('Test video download throttling', function () {
+  const maxBytesPerSecond = 200 * 1024 // 200 KB/s
+
+  async function prepareServer (serverNumber: number, downloadConfig: {
+    max_total_bytes_per_second: number
+    max_bytes_per_ip_per_second: number
+  }) {
+    const server = await createSingleServer(serverNumber, {
+      download: downloadConfig
+    })
+
+    await setAccessTokensToServers([ server ])
+    await setDefaultVideoChannel([ server ])
+    await server.config.enableTranscoding({ hls: true, webVideo: true, resolutions: 'min' })
+
+    // Use a fixture that is big enough to be able to measure throttling
+    const videoId = (await server.videos.quickUpload({ name: 'download-throttle-' + serverNumber, fixture: '60fps_720p_small.mp4' })).uuid
+    await waitJobs([ server ])
+
+    return { server, videoId }
+  }
+
+  async function getClassicWebVideoDownload (server: PeerTubeServer, videoId: string) {
+    const video = await server.videos.get({ id: videoId })
+    const file = video.files.find(f => f.hasVideo === true)
+
+    expect(file).to.exist
+
+    return async () => {
+      const res = await makeRawRequest({
+        url: file.fileDownloadUrl,
+        responseType: 'arraybuffer',
+        expectedStatus: HttpStatusCode.OK_200
+      })
+
+      return res.body as Buffer
+    }
+  }
+
+  async function getGeneratedDownload (server: PeerTubeServer, videoId: string) {
+    const video = await server.videos.get({ id: videoId })
+    const hlsVideoFile = getHLS(video).files.find(f => f.hasVideo === true)
+
+    expect(hlsVideoFile).to.exist
+
+    return () => {
+      return server.videos.generateDownload({
+        videoId,
+        videoFileIds: [ hlsVideoFile.id ]
+      })
+    }
+  }
+
+  async function assertDownloadIsThrottled (download: () => Promise<Buffer>, bytesPerSecond: number) {
+    const start = Date.now()
+    const body = await download()
+    const elapsed = Date.now() - start
+
+    expect(body.length).to.be.greaterThan(0)
+
+    const expectedMinMs = (body.length / bytesPerSecond) * 1000
+    expect(elapsed).to.be.at.least(expectedMinMs * 0.7)
+  }
+
+  async function assertConcurrentDownloadsShareBandwidth (downloads: (() => Promise<Buffer>)[], bytesPerSecond: number) {
+    const start = Date.now()
+    const bodies = await Promise.all(downloads.map(download => download()))
+    const elapsed = Date.now() - start
+
+    const totalBytes = bodies.reduce((sum, body) => sum + body.length, 0)
+    const expectedMinMs = (totalBytes / bytesPerSecond) * 1000
+
+    expect(totalBytes).to.be.greaterThan(0)
+    expect(elapsed).to.be.at.least(expectedMinMs * 0.7)
+  }
+
+  describe('With max_total_bytes_per_second', function () {
+    let server: PeerTubeServer
+    let videoId: string
+
+    before(async function () {
+      this.timeout(120000)
+      ;({ server, videoId } = await prepareServer(1, {
+        max_total_bytes_per_second: maxBytesPerSecond,
+        max_bytes_per_ip_per_second: null
+      }))
+    })
+
+    it('Should throttle classic web video downloads', async function () {
+      this.timeout(120000)
+
+      await assertDownloadIsThrottled(await getClassicWebVideoDownload(server, videoId), maxBytesPerSecond)
+    })
+
+    it('Should throttle generated downloads', async function () {
+      this.timeout(120000)
+
+      await assertDownloadIsThrottled(await getGeneratedDownload(server, videoId), maxBytesPerSecond)
+    })
+
+    it('Should share bandwidth between concurrent download requests', async function () {
+      this.timeout(120000)
+
+      const download = await getClassicWebVideoDownload(server, videoId)
+
+      await assertConcurrentDownloadsShareBandwidth([ download, download ], maxBytesPerSecond)
+    })
+
+    after(async function () {
+      await cleanupTests([ server ])
+    })
+  })
+
+  describe('With max_bytes_per_ip_per_second', function () {
+    let server: PeerTubeServer
+    let videoId: string
+
+    before(async function () {
+      this.timeout(120000)
+      ;({ server, videoId } = await prepareServer(1, {
+        max_total_bytes_per_second: null,
+        max_bytes_per_ip_per_second: maxBytesPerSecond
+      }))
+    })
+
+    it('Should share bandwidth between concurrent requests of the same IP', async function () {
+      this.timeout(120000)
+
+      const download = await getClassicWebVideoDownload(server, videoId)
+
+      await assertConcurrentDownloadsShareBandwidth([ download, download ], maxBytesPerSecond)
+    })
+
+    after(async function () {
+      await cleanupTests([ server ])
+    })
+  })
+})

packages/tests/src/server-helpers/stream-throttle.ts

@@ -0,0 +1,174 @@
+/* oxlint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import { expect } from 'chai'
+import { Readable } from 'stream'
+import { pipeline } from 'stream/promises'
+import { ThrottleStream } from '@peertube/peertube-server/core/helpers/stream-throttle.js'
+
+async function collectStream (readable: Readable): Promise<Buffer> {
+  const chunks: Buffer[] = []
+  for await (const chunk of readable) {
+    chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
+  }
+  return Buffer.concat(chunks)
+}
+
+async function throttledCollect (
+  input: Buffer,
+  options: ConstructorParameters<typeof ThrottleStream>[0]
+): Promise<{ body: Buffer, elapsed: number }> {
+  const throttle = new ThrottleStream(options)
+  const readable = Readable.from([ input ])
+
+  const start = Date.now()
+  const chunks: Buffer[] = []
+
+  await pipeline(readable, throttle, async function (source) {
+    for await (const chunk of source) {
+      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
+    }
+  })
+
+  return { body: Buffer.concat(chunks), elapsed: Date.now() - start }
+}
+
+describe('ThrottleStream', function () {
+  describe('Constructor validation', function () {
+    it('Should throw when neither totalBytesPerSecond nor bytesPerIpPerSecond is provided', function () {
+      expect(() => new ThrottleStream({})).to.throw('At least one throttle speed must be provided')
+    })
+
+    it('Should throw when both values are explicitly undefined', function () {
+      expect(() => new ThrottleStream({ totalBytesPerSecond: undefined, bytesPerIpPerSecond: undefined })).to.throw(
+        'At least one throttle speed must be provided'
+      )
+    })
+
+    it('Should not throw when totalBytesPerSecond is provided', function () {
+      expect(() => new ThrottleStream({ totalBytesPerSecond: 1024 })).to.not.throw()
+    })
+
+    it('Should not throw when bytesPerIpPerSecond is provided', function () {
+      expect(() => new ThrottleStream({ bytesPerIpPerSecond: 1024, ip: '127.0.0.1' })).to.not.throw()
+    })
+
+    it('Should not throw when both are provided', function () {
+      expect(() => new ThrottleStream({ totalBytesPerSecond: 1024, bytesPerIpPerSecond: 512, ip: '127.0.0.1' })).to.not.throw()
+    })
+  })
+
+  describe('Byte preservation', function () {
+    it('Should pass through all bytes unchanged', async function () {
+      const input = Buffer.from('hello world, this is a test of the throttle stream')
+      const { body } = await throttledCollect(input, { totalBytesPerSecond: 1024 * 1024 })
+      expect(body).to.deep.equal(input)
+    })
+
+    it('Should preserve bytes across multiple chunks', async function () {
+      const chunk1 = Buffer.from('first chunk ')
+      const chunk2 = Buffer.from('second chunk')
+      const throttle = new ThrottleStream({ totalBytesPerSecond: 1024 * 1024 })
+      const readable = Readable.from([ chunk1, chunk2 ])
+
+      const chunks: Buffer[] = []
+      await pipeline(readable, throttle, async function (source) {
+        for await (const chunk of source) {
+          chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
+        }
+      })
+
+      const result = Buffer.concat(chunks)
+      expect(result).to.deep.equal(Buffer.concat([ chunk1, chunk2 ]))
+    })
+
+    it('Should handle an empty stream', async function () {
+      const throttle = new ThrottleStream({ totalBytesPerSecond: 1024 })
+      const readable = Readable.from([])
+
+      const body = await collectStream(readable.pipe(throttle))
+      expect(body.length).to.equal(0)
+    })
+
+    it('Should handle a single-byte chunk', async function () {
+      const input = Buffer.from([ 0x42 ])
+      const { body } = await throttledCollect(input, { totalBytesPerSecond: 1024 * 1024 })
+      expect(body).to.deep.equal(input)
+    })
+  })
+
+  describe('Rate enforcement — totalBytesPerSecond', function () {
+    it('Should complete near-instantly when rate is much higher than data size', async function () {
+      const input = Buffer.alloc(1024) // 1 KB
+      const { elapsed } = await throttledCollect(input, { totalBytesPerSecond: 100 * 1024 * 1024 }) // 100 MB/s
+
+      expect(elapsed).to.be.lessThan(500)
+    })
+
+    it('Should take at least the expected time when rate is constrained', async function () {
+      const bytesPerSecond = 10 * 1024 // 10 KB/s
+      const chunkSize = bytesPerSecond / 10 // 10 chunks of 1 KB each
+      const chunks = Array.from({ length: 10 }, () => Buffer.alloc(chunkSize))
+      const totalBytes = chunkSize * chunks.length
+
+      const throttle = new ThrottleStream({ totalBytesPerSecond: bytesPerSecond })
+      const readable = Readable.from(chunks)
+      const out: Buffer[] = []
+
+      const start = Date.now()
+      await pipeline(readable, throttle, async function (source) {
+        for await (const chunk of source) {
+          out.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
+        }
+      })
+      const elapsed = Date.now() - start
+
+      expect(Buffer.concat(out).length).to.equal(totalBytes)
+      // 10 chunks: first passes immediately, remaining 9 each wait ~100ms → ~900ms
+      expect(elapsed).to.be.at.least(700)
+    })
+  })
+
+  describe('Rate enforcement — bytesPerIpPerSecond', function () {
+    it('Should complete near-instantly when rate is much higher than data size', async function () {
+      this.timeout(5000)
+
+      const input = Buffer.alloc(1024)
+      const { elapsed } = await throttledCollect(input, { bytesPerIpPerSecond: 100 * 1024 * 1024, ip: '10.0.0.1' })
+
+      expect(elapsed).to.be.lessThan(500)
+    })
+
+    it('Should take at least the expected time when rate is constrained', async function () {
+      const bytesPerSecond = 10 * 1024 // 10 KB/s
+      const chunkSize = bytesPerSecond / 10 // 10 chunks of 1 KB each
+      const chunks = Array.from({ length: 10 }, () => Buffer.alloc(chunkSize))
+      const totalBytes = chunkSize * chunks.length
+
+      const throttle = new ThrottleStream({ bytesPerIpPerSecond: bytesPerSecond, ip: '10.0.0.2' })
+      const readable = Readable.from(chunks)
+      const out: Buffer[] = []
+
+      const start = Date.now()
+      await pipeline(readable, throttle, async function (source) {
+        for await (const chunk of source) {
+          out.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
+        }
+      })
+      const elapsed = Date.now() - start
+
+      expect(Buffer.concat(out).length).to.equal(totalBytes)
+      // 10 chunks: first passes immediately, remaining 9 each wait ~100ms → ~900ms
+      expect(elapsed).to.be.at.least(700)
+    })
+
+    it('Should not throttle when no IP is provided even with bytesPerIpPerSecond set', async function () {
+      this.timeout(5000)
+
+      // Without an IP the per-IP limiter should not apply
+      const input = Buffer.alloc(50 * 1024)
+      const { elapsed } = await throttledCollect(input, { bytesPerIpPerSecond: 1024 }) // 1 KB/s would be very slow
+
+      expect(elapsed).to.be.lessThan(2000)
+    })
+  })
+})

server/core/controllers/download.ts

@@ -21,7 +21,9 @@ import contentDisposition from 'content-disposition'
 import cors from 'cors'
 import express from 'express'
 import { join } from 'path'
+import { createReadStream } from 'fs'
 import { pipeline } from 'stream/promises'
+import { ThrottleStream } from '@server/helpers/stream-throttle.js'
 import { DOWNLOAD_PATHS, WEBSERVER } from '../initializers/constants.js'
 import {
   asyncMiddleware,
@@ -178,7 +180,7 @@ async function downloadWebVideoFile (req: express.Request, res: express.Response
   }
 
   await VideoPathManager.Instance.makeAvailableVideoFile(videoFile.withVideoOrPlaylist(video), path => {
-    return res.download(path, downloadFilename)
+    return downloadLocalFileWithOptionalThrottle({ res, path, downloadFilename, ip: req.ip })
   })
 }
 
@@ -221,7 +223,7 @@ async function downloadHLSVideoFile (req: express.Request, res: express.Response
   }
 
   await VideoPathManager.Instance.makeAvailableVideoFile(videoFile.withVideoOrPlaylist(streamingPlaylist), path => {
-    return res.download(path, downloadFilename)
+    return downloadLocalFileWithOptionalThrottle({ res, path, downloadFilename, ip: req.ip })
   })
 }
 
@@ -291,7 +293,11 @@ async function downloadGeneratedVideoFile (req: express.Request, res: express.Re
     .catch(err => logger.error(`Cannot process local download stats for video ${video.uuid}`, { err, ...lTags(video.uuid) }))
 
   try {
-    await new VideoDownload({ video, videoFiles }).muxToMergeVideoFiles(res)
+    await new VideoDownload({ video, videoFiles }).muxToMergeVideoFiles(res, {
+      totalBytesPerSecond: CONFIG.DOWNLOAD.MAX_TOTAL_BYTES_PER_SECOND,
+      bytesPerIpPerSecond: CONFIG.DOWNLOAD.MAX_BYTES_PER_IP_PER_SECOND,
+      ip: req.ip
+    })
   } catch (err) {
     // muxToMergeVideoFiles has already logged the error
     res.fail({
@@ -313,8 +319,12 @@ function downloadUserExport (req: express.Request, res: express.Response) {
     return redirectUserExportToObjectStorage({ res, userExport, downloadFilename })
   }
 
-  res.download(getFSUserExportFilePath(userExport), downloadFilename)
-  return Promise.resolve()
+  return downloadLocalFileWithOptionalThrottle({
+    res,
+    path: getFSUserExportFilePath(userExport),
+    downloadFilename,
+    ip: req.ip
+  })
 }
 
 function downloadOriginalFile (req: express.Request, res: express.Response) {
@@ -326,8 +336,12 @@ function downloadOriginalFile (req: express.Request, res: express.Response) {
     return redirectOriginalFileToObjectStorage({ res, videoSource, downloadFilename })
   }
 
-  res.download(VideoPathManager.Instance.getFSOriginalVideoFilePath(videoSource.keptOriginalFilename), downloadFilename)
-  return Promise.resolve()
+  return downloadLocalFileWithOptionalThrottle({
+    res,
+    path: VideoPathManager.Instance.getFSOriginalVideoFilePath(videoSource.keptOriginalFilename),
+    downloadFilename,
+    ip: req.ip
+  })
 }
 
 // ---------------------------------------------------------------------------
@@ -386,6 +400,25 @@ function checkAllowResult (res: express.Response, allowParameters: any, result?:
   return true
 }
 
+async function downloadLocalFileWithOptionalThrottle (options: {
+  res: express.Response
+  path: string
+  downloadFilename: string
+  ip?: string
+}) {
+  const { res, path, downloadFilename, ip } = options
+
+  const totalBytesPerSecond = CONFIG.DOWNLOAD.MAX_TOTAL_BYTES_PER_SECOND
+  const bytesPerIpPerSecond = CONFIG.DOWNLOAD.MAX_BYTES_PER_IP_PER_SECOND
+
+  if (!totalBytesPerSecond && !bytesPerIpPerSecond) return res.download(path, downloadFilename)
+
+  res.setHeader('Content-Disposition', contentDisposition(encodeURI(downloadFilename)))
+  res.setHeader('Content-Type', 'application/octet-stream')
+
+  await pipeline(createReadStream(path), new ThrottleStream({ totalBytesPerSecond, bytesPerIpPerSecond, ip }), res)
+}
+
 async function redirectVideoDownloadToObjectStorage (options: {
   res: express.Response
   video: MVideo

server/core/helpers/stream-throttle.ts

@@ -0,0 +1,114 @@
+import { exists } from '@peertube/peertube-core-utils'
+import { Transform, TransformCallback } from 'stream'
+
+type SharedThrottleOptions = {
+  totalBytesPerSecond?: number
+  bytesPerIpPerSecond?: number
+  ip?: string
+}
+
+class SharedThrottleState {
+  private readonly ipNextAvailableAt = new Map<string, number>()
+  private totalNextAvailableAt = 0
+  private lastCleanup = 0
+
+  async throttle (
+    options: SharedThrottleOptions & {
+      bytes: number
+    }
+  ) {
+    const { bytes, totalBytesPerSecond, bytesPerIpPerSecond, ip } = options
+
+    const now = Date.now()
+
+    const totalReadyAt = totalBytesPerSecond
+      ? Math.max(now, this.totalNextAvailableAt)
+      : now
+
+    const ipReadyAt = bytesPerIpPerSecond && ip
+      ? Math.max(now, this.ipNextAvailableAt.get(ip) ?? 0)
+      : now
+
+    const readyAt = Math.max(totalReadyAt, ipReadyAt)
+
+    if (totalBytesPerSecond) {
+      this.totalNextAvailableAt = readyAt + this.computeDurationMs(bytes, totalBytesPerSecond)
+    }
+
+    if (bytesPerIpPerSecond && ip) {
+      this.ipNextAvailableAt.set(ip, readyAt + this.computeDurationMs(bytes, bytesPerIpPerSecond))
+    }
+
+    this.cleanup(now)
+
+    const delay = readyAt - now
+    if (delay <= 0) return
+
+    await new Promise<void>(resolve => setTimeout(resolve, delay))
+  }
+
+  private computeDurationMs (bytes: number, bytesPerSecond: number) {
+    return Math.ceil((bytes / bytesPerSecond) * 1000)
+  }
+
+  private cleanup (now: number) {
+    if (now - this.lastCleanup < 60_000) return
+
+    this.lastCleanup = now
+
+    for (const [ ip, nextAvailableAt ] of this.ipNextAvailableAt) {
+      if (nextAvailableAt < now - 60_000) {
+        this.ipNextAvailableAt.delete(ip)
+      }
+    }
+  }
+}
+
+const sharedThrottleState = new SharedThrottleState()
+
+/**
+ * A Transform stream that throttles throughput to a given number of bytes per second.
+ */
+export class ThrottleStream extends Transform {
+  private readonly totalBytesPerSecond?: number
+  private readonly bytesPerIpPerSecond?: number
+  private readonly ip?: string
+
+  constructor (options: SharedThrottleOptions) {
+    super()
+
+    if (!exists(options.totalBytesPerSecond) && !exists(options.bytesPerIpPerSecond)) {
+      throw new Error('At least one throttle speed must be provided')
+    }
+
+    if (exists(options.bytesPerIpPerSecond) && !exists(options.ip)) {
+      throw new Error('An ip must be provided when bytesPerIpPerSecond is set')
+    }
+    this.totalBytesPerSecond = options.totalBytesPerSecond
+    this.bytesPerIpPerSecond = options.bytesPerIpPerSecond
+    this.ip = options.ip
+  }
+
+  _transform (chunk: Buffer, _encoding: BufferEncoding, done: TransformCallback) {
+    this.handleChunk(chunk)
+      .then(() => done())
+      .catch(done)
+  }
+
+  _flush (done: TransformCallback) {
+    return done()
+  }
+
+  private async handleChunk (chunk: Buffer) {
+    if (this.totalBytesPerSecond || this.bytesPerIpPerSecond) {
+      await sharedThrottleState.throttle({
+        bytes: chunk.length,
+        totalBytesPerSecond: this.totalBytesPerSecond,
+        bytesPerIpPerSecond: this.bytesPerIpPerSecond,
+        ip: this.ip
+      })
+    }
+
+    this.push(chunk)
+  }
+}

server/core/initializers/checker-before-init.ts

@@ -264,6 +264,8 @@ export function checkMissedConfig () {
     'storyboards.enabled',
     'webrtc.stun_servers',
     'nsfw_flags_settings.enabled',
+    'download.max_total_bytes_per_second',
+    'download.max_bytes_per_ip_per_second',
     'download_generate_video.max_parallel_downloads',
     'video_comments.accept_remote_comments'
   ]

server/core/initializers/config.ts

@@ -124,6 +124,15 @@ const CONFIG = {
     ENABLED: config.get<boolean>('nsfw_flags_settings.enabled')
   },
 
+  DOWNLOAD: {
+    MAX_TOTAL_BYTES_PER_SECOND: config.get<string | number | null>('download.max_total_bytes_per_second') === null
+      ? null
+      : parseBytes(config.get<string | number>('download.max_total_bytes_per_second')),
+    MAX_BYTES_PER_IP_PER_SECOND: config.get<string | number | null>('download.max_bytes_per_ip_per_second') === null
+      ? null
+      : parseBytes(config.get<string | number>('download.max_bytes_per_ip_per_second'))
+  },
+
   DOWNLOAD_GENERATE_VIDEO: {
     MAX_PARALLEL_DOWNLOADS: config.get<number>('download_generate_video.max_parallel_downloads')
   },

server/core/lib/video-download.ts

@@ -3,11 +3,12 @@ import { FileStorage } from '@peertube/peertube-models'
 import { getFFmpegCommandWrapperOptions } from '@server/helpers/ffmpeg/ffmpeg-options.js'
 import { logger } from '@server/helpers/logger.js'
 import { buildRequestError, doRequestAndSaveToFile, generateRequestStream } from '@server/helpers/requests.js'
+import { ThrottleStream } from '@server/helpers/stream-throttle.js'
 import { REQUEST_TIMEOUTS } from '@server/initializers/constants.js'
 import { isWebVideoFile, MVideoFile, MVideoThumbnails } from '@server/types/models/index.js'
 import { createReadStream } from 'fs'
 import { remove } from 'fs-extra/esm'
-import { Readable, Writable } from 'stream'
+import { PassThrough, Readable, Writable } from 'stream'
 import { pipeline } from 'stream/promises'
 import { lTags } from './object-storage/shared/index.js'
 import {
@@ -39,8 +40,16 @@ export class VideoDownload {
     this.videoFiles = options.videoFiles
   }
 
-  async muxToMergeVideoFiles (output: Writable) {
+  async muxToMergeVideoFiles (output: Writable, options?: {
+    totalBytesPerSecond: number
+    bytesPerIpPerSecond: number
+    ip: string
+  }) {
     return new Promise<void>(async (res, rej) => {
+      const totalBytesPerSecond = options?.totalBytesPerSecond
+      const bytesPerIpPerSecond = options?.bytesPerIpPerSecond
+      const ip = options?.ip
+
       try {
         VideoDownload.totalDownloads++
 
@@ -63,22 +72,40 @@ export class VideoDownload {
             ? createReadStream(this.inputs[0])
             : this.inputs[0]
 
-          await pipeline(input, output)
+          const throttleStream = totalBytesPerSecond || bytesPerIpPerSecond
+            ? new ThrottleStream({ totalBytesPerSecond, bytesPerIpPerSecond, ip })
+            : new PassThrough()
+
+          await pipeline(input, throttleStream, output)
+
+          res()
         } else {
           logger.info(`Muxing files for video ${this.video.url}`, { inputs: this.inputsToLog(), ...lTags(this.video.uuid) })
 
           this.ffmpegContainer = new FFmpegContainer(getFFmpegCommandWrapperOptions('vod'))
 
+          const throttleStream = totalBytesPerSecond || bytesPerIpPerSecond
+            ? new ThrottleStream({ totalBytesPerSecond, bytesPerIpPerSecond, ip })
+            : undefined
+
+          const finalOutput = throttleStream ?? output
+
+          const throttlePipeline = throttleStream
+            ? pipeline(throttleStream, output)
+            : undefined
+
           try {
             await this.ffmpegContainer.mergeInputs({
               inputs: this.inputs,
-              output,
+              output: finalOutput,
               logError: false,
 
               // Include a cover if this is an audio file
               coverPath
             })
 
+            if (throttleStream) await throttlePipeline
+
             logger.info(`Mux ended for video ${this.video.url}`, { inputs: this.inputsToLog(), ...lTags(this.video.uuid) })
 
             res()