Live supports object storage

 * Sync live files (segments, master playlist, resolution playlist,
   segment sha file) into object storage
 * Automatically delete them when the live ends
 * Segment sha file is now a file on disk, and not stored in memory
   anymore

5 files changed, 270 insertions, 131 deletions

diff --git a/server/tests/api/live/live-fast-restream.ts b/server/tests/api/live/live-fast-restream.ts
index 502959258..3ea6be9ff 100644
--- a/server/tests/api/live/live-fast-restream.ts
+++ b/server/tests/api/live/live-fast-restream.ts
@@ -59,7 +59,7 @@ describe('Fast restream in live', function () {
       const video = await server.videos.get({ id: liveId })
       expect(video.streamingPlaylists).to.have.lengthOf(1)
 
-      await server.live.getSegment({ videoUUID: liveId, segment: 0, playlistNumber: 0 })
+      await server.live.getSegmentFile({ videoUUID: liveId, segment: 0, playlistNumber: 0 })
       await makeRawRequest(video.streamingPlaylists[0].playlistUrl, HttpStatusCode.OK_200)
       await makeRawRequest(video.streamingPlaylists[0].segmentsSha256Url, HttpStatusCode.OK_200)
 
diff --git a/server/tests/api/live/live.ts b/server/tests/api/live/live.ts
index 4e070832d..5dd2bd9ab 100644
--- a/server/tests/api/live/live.ts
+++ b/server/tests/api/live/live.ts
@@ -3,7 +3,7 @@
 import { expect } from 'chai'
 import { basename, join } from 'path'
 import { ffprobePromise, getVideoStream } from '@server/helpers/ffmpeg'
-import { checkLiveSegmentHash, checkResolutionsInMasterPlaylist, testImage } from '@server/tests/shared'
+import { testImage, testVideoResolutions } from '@server/tests/shared'
 import { getAllFiles, wait } from '@shared/core-utils'
 import {
   HttpStatusCode,
@@ -372,46 +372,6 @@ describe('Test live', function () {
       return uuid
     }
 
-    async function testVideoResolutions (liveVideoId: string, resolutions: number[]) {
-      for (const server of servers) {
-        const { data } = await server.videos.list()
-        expect(data.find(v => v.uuid === liveVideoId)).to.exist
-
-        const video = await server.videos.get({ id: liveVideoId })
-
-        expect(video.streamingPlaylists).to.have.lengthOf(1)
-
-        const hlsPlaylist = video.streamingPlaylists.find(s => s.type === VideoStreamingPlaylistType.HLS)
-        expect(hlsPlaylist).to.exist
-
-        // Only finite files are displayed
-        expect(hlsPlaylist.files).to.have.lengthOf(0)
-
-        await checkResolutionsInMasterPlaylist({ server, playlistUrl: hlsPlaylist.playlistUrl, resolutions })
-
-        for (let i = 0; i < resolutions.length; i++) {
-          const segmentNum = 3
-          const segmentName = `${i}-00000${segmentNum}.ts`
-          await commands[0].waitUntilSegmentGeneration({ videoUUID: video.uuid, playlistNumber: i, segment: segmentNum })
-
-          const subPlaylist = await servers[0].streamingPlaylists.get({
-            url: `${servers[0].url}/static/streaming-playlists/hls/${video.uuid}/${i}.m3u8`
-          })
-
-          expect(subPlaylist).to.contain(segmentName)
-
-          const baseUrlAndPath = servers[0].url + '/static/streaming-playlists/hls'
-          await checkLiveSegmentHash({
-            server,
-            baseUrlSegment: baseUrlAndPath,
-            videoUUID: video.uuid,
-            segmentName,
-            hlsPlaylist
-          })
-        }
-      }
-    }
-
     function updateConf (resolutions: number[]) {
       return servers[0].config.updateCustomSubConfig({
         newConfig: {
@@ -449,7 +409,14 @@ describe('Test live', function () {
       await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
       await waitJobs(servers)
 
-      await testVideoResolutions(liveVideoId, [ 720 ])
+      await testVideoResolutions({
+        originServer: servers[0],
+        servers,
+        liveVideoId,
+        resolutions: [ 720 ],
+        objectStorage: false,
+        transcoded: true
+      })
 
       await stopFfmpeg(ffmpegCommand)
     })
@@ -477,7 +444,14 @@ describe('Test live', function () {
       await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
       await waitJobs(servers)
 
-      await testVideoResolutions(liveVideoId, resolutions.concat([ 720 ]))
+      await testVideoResolutions({
+        originServer: servers[0],
+        servers,
+        liveVideoId,
+        resolutions: resolutions.concat([ 720 ]),
+        objectStorage: false,
+        transcoded: true
+      })
 
       await stopFfmpeg(ffmpegCommand)
     })
@@ -522,7 +496,14 @@ describe('Test live', function () {
       await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
       await waitJobs(servers)
 
-      await testVideoResolutions(liveVideoId, resolutions)
+      await testVideoResolutions({
+        originServer: servers[0],
+        servers,
+        liveVideoId,
+        resolutions,
+        objectStorage: false,
+        transcoded: true
+      })
 
       await stopFfmpeg(ffmpegCommand)
       await commands[0].waitUntilEnded({ videoId: liveVideoId })
@@ -611,7 +592,14 @@ describe('Test live', function () {
       await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
       await waitJobs(servers)
 
-      await testVideoResolutions(liveVideoId, resolutions)
+      await testVideoResolutions({
+        originServer: servers[0],
+        servers,
+        liveVideoId,
+        resolutions,
+        objectStorage: false,
+        transcoded: true
+      })
 
       await stopFfmpeg(ffmpegCommand)
       await commands[0].waitUntilEnded({ videoId: liveVideoId })
@@ -640,7 +628,14 @@ describe('Test live', function () {
       await waitUntilLivePublishedOnAllServers(servers, liveVideoId)
       await waitJobs(servers)
 
-      await testVideoResolutions(liveVideoId, [ 720 ])
+      await testVideoResolutions({
+        originServer: servers[0],
+        servers,
+        liveVideoId,
+        resolutions: [ 720 ],
+        objectStorage: false,
+        transcoded: true
+      })
 
       await stopFfmpeg(ffmpegCommand)
       await commands[0].waitUntilEnded({ videoId: liveVideoId })
diff --git a/server/tests/api/object-storage/live.ts b/server/tests/api/object-storage/live.ts
index 0958ffe0f..7e16b4c89 100644
--- a/server/tests/api/object-storage/live.ts
+++ b/server/tests/api/object-storage/live.ts
@@ -1,9 +1,9 @@
 /* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
 
 import { expect } from 'chai'
-import { expectStartWith } from '@server/tests/shared'
+import { expectStartWith, testVideoResolutions } from '@server/tests/shared'
 import { areObjectStorageTestsDisabled } from '@shared/core-utils'
-import { HttpStatusCode, LiveVideoCreate, VideoFile, VideoPrivacy } from '@shared/models'
+import { HttpStatusCode, LiveVideoCreate, VideoPrivacy } from '@shared/models'
 import {
   createMultipleServers,
   doubleFollow,
@@ -35,41 +35,43 @@ async function createLive (server: PeerTubeServer, permanent: boolean) {
   return uuid
 }
 
-async function checkFiles (files: VideoFile[]) {
-  for (const file of files) {
-    expectStartWith(file.fileUrl, ObjectStorageCommand.getPlaylistBaseUrl())
+async function checkFilesExist (servers: PeerTubeServer[], videoUUID: string, numberOfFiles: number) {
+  for (const server of servers) {
+    const video = await server.videos.get({ id: videoUUID })
 
-    await makeRawRequest(file.fileUrl, HttpStatusCode.OK_200)
-  }
-}
+    expect(video.files).to.have.lengthOf(0)
+    expect(video.streamingPlaylists).to.have.lengthOf(1)
 
-async function getFiles (server: PeerTubeServer, videoUUID: string) {
-  const video = await server.videos.get({ id: videoUUID })
+    const files = video.streamingPlaylists[0].files
+    expect(files).to.have.lengthOf(numberOfFiles)
 
-  expect(video.files).to.have.lengthOf(0)
-  expect(video.streamingPlaylists).to.have.lengthOf(1)
+    for (const file of files) {
+      expectStartWith(file.fileUrl, ObjectStorageCommand.getPlaylistBaseUrl())
 
-  return video.streamingPlaylists[0].files
+      await makeRawRequest(file.fileUrl, HttpStatusCode.OK_200)
+    }
+  }
 }
 
-async function streamAndEnd (servers: PeerTubeServer[], liveUUID: string) {
-  const ffmpegCommand = await servers[0].live.sendRTMPStreamInVideo({ videoId: liveUUID })
-  await waitUntilLivePublishedOnAllServers(servers, liveUUID)
-
-  const videoLiveDetails = await servers[0].videos.get({ id: liveUUID })
-  const liveDetails = await servers[0].live.get({ videoId: liveUUID })
+async function checkFilesCleanup (server: PeerTubeServer, videoUUID: string, resolutions: number[]) {
+  const resolutionFiles = resolutions.map((_value, i) => `${i}.m3u8`)
 
-  await stopFfmpeg(ffmpegCommand)
-
-  if (liveDetails.permanentLive) {
-    await waitUntilLiveWaitingOnAllServers(servers, liveUUID)
-  } else {
-    await waitUntilLiveReplacedByReplayOnAllServers(servers, liveUUID)
+  for (const playlistName of [ 'master.m3u8' ].concat(resolutionFiles)) {
+    await server.live.getPlaylistFile({
+      videoUUID,
+      playlistName,
+      expectedStatus: HttpStatusCode.NOT_FOUND_404,
+      objectStorage: true
+    })
   }
 
-  await waitJobs(servers)
-
-  return { videoLiveDetails, liveDetails }
+  await server.live.getSegmentFile({
+    videoUUID,
+    playlistNumber: 0,
+    segment: 0,
+    objectStorage: true,
+    expectedStatus: HttpStatusCode.NOT_FOUND_404
+  })
 }
 
 describe('Object storage for lives', function () {
@@ -100,57 +102,124 @@ describe('Object storage for lives', function () {
       videoUUID = await createLive(servers[0], false)
     })
 
-    it('Should create a live and save the replay on object storage', async function () {
+    it('Should create a live and publish it on object storage', async function () {
+      this.timeout(220000)
+
+      const ffmpegCommand = await servers[0].live.sendRTMPStreamInVideo({ videoId: videoUUID })
+      await waitUntilLivePublishedOnAllServers(servers, videoUUID)
+
+      await testVideoResolutions({
+        originServer: servers[0],
+        servers,
+        liveVideoId: videoUUID,
+        resolutions: [ 720 ],
+        transcoded: false,
+        objectStorage: true
+      })
+
+      await stopFfmpeg(ffmpegCommand)
+    })
+
+    it('Should have saved the replay on object storage', async function () {
       this.timeout(220000)
 
-      await streamAndEnd(servers, videoUUID)
+      await waitUntilLiveReplacedByReplayOnAllServers(servers, videoUUID)
+      await waitJobs(servers)
 
-      for (const server of servers) {
-        const files = await getFiles(server, videoUUID)
-        expect(files).to.have.lengthOf(1)
+      await checkFilesExist(servers, videoUUID, 1)
+    })
 
-        await checkFiles(files)
-      }
+    it('Should have cleaned up live files from object storage', async function () {
+      await checkFilesCleanup(servers[0], videoUUID, [ 720 ])
     })
   })
 
   describe('With live transcoding', async function () {
-    let videoUUIDPermanent: string
-    let videoUUIDNonPermanent: string
+    const resolutions = [ 720, 480, 360, 240, 144 ]
 
     before(async function () {
       await servers[0].config.enableLive({ transcoding: true })
-
-      videoUUIDPermanent = await createLive(servers[0], true)
-      videoUUIDNonPermanent = await createLive(servers[0], false)
     })
 
-    it('Should create a live and save the replay on object storage', async function () {
-      this.timeout(240000)
+    describe('Normal replay', function () {
+      let videoUUIDNonPermanent: string
+
+      before(async function () {
+        videoUUIDNonPermanent = await createLive(servers[0], false)
+      })
+
+      it('Should create a live and publish it on object storage', async function () {
+        this.timeout(240000)
+
+        const ffmpegCommand = await servers[0].live.sendRTMPStreamInVideo({ videoId: videoUUIDNonPermanent })
+        await waitUntilLivePublishedOnAllServers(servers, videoUUIDNonPermanent)
+
+        await testVideoResolutions({
+          originServer: servers[0],
+          servers,
+          liveVideoId: videoUUIDNonPermanent,
+          resolutions,
+          transcoded: true,
+          objectStorage: true
+        })
+
+        await stopFfmpeg(ffmpegCommand)
+      })
 
-      await streamAndEnd(servers, videoUUIDNonPermanent)
+      it('Should have saved the replay on object storage', async function () {
+        this.timeout(220000)
 
-      for (const server of servers) {
-        const files = await getFiles(server, videoUUIDNonPermanent)
-        expect(files).to.have.lengthOf(5)
+        await waitUntilLiveReplacedByReplayOnAllServers(servers, videoUUIDNonPermanent)
+        await waitJobs(servers)
 
-        await checkFiles(files)
-      }
+        await checkFilesExist(servers, videoUUIDNonPermanent, 5)
+      })
+
+      it('Should have cleaned up live files from object storage', async function () {
+        await checkFilesCleanup(servers[0], videoUUIDNonPermanent, resolutions)
+      })
     })
 
-    it('Should create a live and save the replay of permanent live on object storage', async function () {
-      this.timeout(240000)
+    describe('Permanent replay', function () {
+      let videoUUIDPermanent: string
+
+      before(async function () {
+        videoUUIDPermanent = await createLive(servers[0], true)
+      })
+
+      it('Should create a live and publish it on object storage', async function () {
+        this.timeout(240000)
+
+        const ffmpegCommand = await servers[0].live.sendRTMPStreamInVideo({ videoId: videoUUIDPermanent })
+        await waitUntilLivePublishedOnAllServers(servers, videoUUIDPermanent)
+
+        await testVideoResolutions({
+          originServer: servers[0],
+          servers,
+          liveVideoId: videoUUIDPermanent,
+          resolutions,
+          transcoded: true,
+          objectStorage: true
+        })
+
+        await stopFfmpeg(ffmpegCommand)
+      })
+
+      it('Should have saved the replay on object storage', async function () {
+        this.timeout(220000)
 
-      const { videoLiveDetails } = await streamAndEnd(servers, videoUUIDPermanent)
+        await waitUntilLiveWaitingOnAllServers(servers, videoUUIDPermanent)
+        await waitJobs(servers)
 
-      const replay = await findExternalSavedVideo(servers[0], videoLiveDetails)
+        const videoLiveDetails = await servers[0].videos.get({ id: videoUUIDPermanent })
+        const replay = await findExternalSavedVideo(servers[0], videoLiveDetails)
 
-      for (const server of servers) {
-        const files = await getFiles(server, replay.uuid)
-        expect(files).to.have.lengthOf(5)
+        await checkFilesExist(servers, replay.uuid, 5)
+      })
 
-        await checkFiles(files)
-      }
+      it('Should have cleaned up live files from object storage', async function () {
+        await checkFilesCleanup(servers[0], videoUUIDPermanent, resolutions)
+      })
     })
   })
 
diff --git a/server/tests/shared/live.ts b/server/tests/shared/live.ts
index 4bd4786fc..aa79622cb 100644
--- a/server/tests/shared/live.ts
+++ b/server/tests/shared/live.ts
@@ -3,39 +3,92 @@
 import { expect } from 'chai'
 import { pathExists, readdir } from 'fs-extra'
 import { join } from 'path'
-import { LiveVideo } from '@shared/models'
-import { PeerTubeServer } from '@shared/server-commands'
+import { wait } from '@shared/core-utils'
+import { LiveVideo, VideoStreamingPlaylistType } from '@shared/models'
+import { ObjectStorageCommand, PeerTubeServer } from '@shared/server-commands'
+import { checkLiveSegmentHash, checkResolutionsInMasterPlaylist } from './streaming-playlists'
 
 async function checkLiveCleanup (server: PeerTubeServer, videoUUID: string, savedResolutions: number[] = []) {
-  let live: LiveVideo
-
-  try {
-    live = await server.live.get({ videoId: videoUUID })
-  } catch {}
-
   const basePath = server.servers.buildDirectory('streaming-playlists')
   const hlsPath = join(basePath, 'hls', videoUUID)
 
   if (savedResolutions.length === 0) {
+    return checkUnsavedLiveCleanup(server, videoUUID, hlsPath)
+  }
+
+  return checkSavedLiveCleanup(hlsPath, savedResolutions)
+}
+
+// ---------------------------------------------------------------------------
 
-    if (live?.permanentLive) {
-      expect(await pathExists(hlsPath)).to.be.true
+async function testVideoResolutions (options: {
+  originServer: PeerTubeServer
+  servers: PeerTubeServer[]
+  liveVideoId: string
+  resolutions: number[]
+  transcoded: boolean
+  objectStorage: boolean
+}) {
+  const { originServer, servers, liveVideoId, resolutions, transcoded, objectStorage } = options
 
-      const hlsFiles = await readdir(hlsPath)
-      expect(hlsFiles).to.have.lengthOf(1) // Only replays directory
+  for (const server of servers) {
+    const { data } = await server.videos.list()
+    expect(data.find(v => v.uuid === liveVideoId)).to.exist
 
-      const replayDir = join(hlsPath, 'replay')
-      expect(await pathExists(replayDir)).to.be.true
+    const video = await server.videos.get({ id: liveVideoId })
+    expect(video.streamingPlaylists).to.have.lengthOf(1)
 
-      const replayFiles = await readdir(join(hlsPath, 'replay'))
-      expect(replayFiles).to.have.lengthOf(0)
-    } else {
-      expect(await pathExists(hlsPath)).to.be.false
+    const hlsPlaylist = video.streamingPlaylists.find(s => s.type === VideoStreamingPlaylistType.HLS)
+    expect(hlsPlaylist).to.exist
+    expect(hlsPlaylist.files).to.have.lengthOf(0) // Only fragmented mp4 files are displayed
+
+    await checkResolutionsInMasterPlaylist({ server, playlistUrl: hlsPlaylist.playlistUrl, resolutions, transcoded })
+
+    if (objectStorage) {
+      expect(hlsPlaylist.playlistUrl).to.contain(ObjectStorageCommand.getPlaylistBaseUrl())
     }
 
-    return
+    for (let i = 0; i < resolutions.length; i++) {
+      const segmentNum = 3
+      const segmentName = `${i}-00000${segmentNum}.ts`
+      await originServer.live.waitUntilSegmentGeneration({ videoUUID: video.uuid, playlistNumber: i, segment: segmentNum })
+
+      const baseUrl = objectStorage
+        ? ObjectStorageCommand.getPlaylistBaseUrl() + 'hls'
+        : originServer.url + '/static/streaming-playlists/hls'
+
+      if (objectStorage) {
+        // Playlist file upload
+        await wait(500)
+
+        expect(hlsPlaylist.segmentsSha256Url).to.contain(ObjectStorageCommand.getPlaylistBaseUrl())
+      }
+
+      const subPlaylist = await originServer.streamingPlaylists.get({ url: `${baseUrl}/${video.uuid}/${i}.m3u8` })
+
+      expect(subPlaylist).to.contain(segmentName)
+
+      await checkLiveSegmentHash({
+        server,
+        baseUrlSegment: baseUrl,
+        videoUUID: video.uuid,
+        segmentName,
+        hlsPlaylist
+      })
+    }
   }
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  checkLiveCleanup,
+  testVideoResolutions
+}
 
+// ---------------------------------------------------------------------------
+
+async function checkSavedLiveCleanup (hlsPath: string, savedResolutions: number[] = []) {
   const files = await readdir(hlsPath)
 
   // fragmented file and playlist per resolution + master playlist + segments sha256 json file
@@ -56,6 +109,27 @@ async function checkLiveCleanup (server: PeerTubeServer, videoUUID: string, save
   expect(shaFile).to.exist
 }
 
-export {
-  checkLiveCleanup
+async function checkUnsavedLiveCleanup (server: PeerTubeServer, videoUUID: string, hlsPath: string) {
+  let live: LiveVideo
+
+  try {
+    live = await server.live.get({ videoId: videoUUID })
+  } catch {}
+
+  if (live?.permanentLive) {
+    expect(await pathExists(hlsPath)).to.be.true
+
+    const hlsFiles = await readdir(hlsPath)
+    expect(hlsFiles).to.have.lengthOf(1) // Only replays directory
+
+    const replayDir = join(hlsPath, 'replay')
+    expect(await pathExists(replayDir)).to.be.true
+
+    const replayFiles = await readdir(join(hlsPath, 'replay'))
+    expect(replayFiles).to.have.lengthOf(0)
+
+    return
+  }
+
+  expect(await pathExists(hlsPath)).to.be.false
 }
diff --git a/server/tests/shared/streaming-playlists.ts b/server/tests/shared/streaming-playlists.ts
index 4d82b3654..eff34944b 100644
--- a/server/tests/shared/streaming-playlists.ts
+++ b/server/tests/shared/streaming-playlists.ts
@@ -26,7 +26,7 @@ async function checkSegmentHash (options: {
   const offset = parseInt(matches[2], 10)
   const range = `${offset}-${offset + length - 1}`
 
-  const segmentBody = await command.getSegment({
+  const segmentBody = await command.getFragmentedSegment({
     url: `${baseUrlSegment}/${videoName}`,
     expectedStatus: HttpStatusCode.PARTIAL_CONTENT_206,
     range: `bytes=${range}`
@@ -46,7 +46,7 @@ async function checkLiveSegmentHash (options: {
   const { server, baseUrlSegment, videoUUID, segmentName, hlsPlaylist } = options
   const command = server.streamingPlaylists
 
-  const segmentBody = await command.getSegment({ url: `${baseUrlSegment}/${videoUUID}/${segmentName}` })
+  const segmentBody = await command.getFragmentedSegment({ url: `${baseUrlSegment}/${videoUUID}/${segmentName}` })
   const shaBody = await command.getSegmentSha256({ url: hlsPlaylist.segmentsSha256Url })
 
   expect(sha256(segmentBody)).to.equal(shaBody[segmentName])
@@ -56,15 +56,16 @@ async function checkResolutionsInMasterPlaylist (options: {
   server: PeerTubeServer
   playlistUrl: string
   resolutions: number[]
+  transcoded?: boolean // default true
 }) {
-  const { server, playlistUrl, resolutions } = options
+  const { server, playlistUrl, resolutions, transcoded = true } = options
 
   const masterPlaylist = await server.streamingPlaylists.get({ url: playlistUrl })
 
   for (const resolution of resolutions) {
-    const reg = new RegExp(
-      '#EXT-X-STREAM-INF:BANDWIDTH=\\d+,RESOLUTION=\\d+x' + resolution + ',(FRAME-RATE=\\d+,)?CODECS="avc1.64001f,mp4a.40.2"'
-    )
+    const reg = transcoded
+      ? new RegExp('#EXT-X-STREAM-INF:BANDWIDTH=\\d+,RESOLUTION=\\d+x' + resolution + ',(FRAME-RATE=\\d+,)?CODECS="avc1.64001f,mp4a.40.2"')
+      : new RegExp('#EXT-X-STREAM-INF:BANDWIDTH=\\d+,RESOLUTION=\\d+x' + resolution + '')
 
     expect(masterPlaylist).to.match(reg)
   }