Fix MBS Broadcast content delivery crashes and Distribution Session bugs

README.md

@@ -1,15 +1,58 @@
-<h1 align="center">5G MBS User Services: MBS Transport Function</h1>
 <p align="center">
-  <img src="https://img.shields.io/github/v/tag/5G-MAG/rt-mbs-transport-function?label=version" alt="Version">
-  <img src="https://img.shields.io/badge/Status-Under_Development-yellow" alt="Under Development">
-  <img src="https://img.shields.io/badge/License-5G--MAG%20Public%20License%20(v1.0)-blue" alt="License">
+  <img src=".github/banner.svg" width="100%" alt="5G-MAG Reference Tools, 5G Multicast Broadcast Services: MBS Transport Function (MBSTF)">
 </p>
 
+<p align="center">
+  The MBS Transport Function (MBSTF), the user-plane function that delivers MBS content as
+  FLUTE object streams, per 3GPP TS 26.502 and TS 29.581.
+</p>
+
+<p align="center">
+  <img alt="Status: under development"
+    src="https://img.shields.io/badge/Status-Under_Development-yellow">
+  <a href="https://github.com/5G-MAG/rt-mbs-transport-function/releases"><img alt="Version"
+    src="https://img.shields.io/github/v/release/5G-MAG/rt-mbs-transport-function?label=Version&sort=semver"></a>
+  <a href="LICENSE"><img alt="5G-MAG Public License v1.0"
+    src="https://img.shields.io/badge/License-5G--MAG%20PL%20v1.0-blue"></a>
+</p>
+
+<p align="center">
+  <a href="https://www.5g-mag.com/reference-tools/5g-multicast-broadcast-services">Project page</a> &nbsp;&middot;&nbsp;
+  <a href="https://github.com/5G-MAG/rt-mbs-transport-function/issues">Issues</a> &nbsp;&middot;&nbsp;
+  <a href="https://www.5g-mag.com/contributing">Contributing</a>
+</p>
+
+---
+
+## At a glance
+
+|  |  |
+|---|---|
+| **Implements** | 3GPP TS 26.502, *5G multicast-broadcast services; User Service architecture*, and TS 29.581, *Nmb8 service API* |
+| **Role** | MBSTF: object ingest, FLUTE packaging, MBS delivery over the user plane |
+| **Built with** | C++ and meson, on top of Open5GS |
+| **Works with** | [rt-mbs-function](https://github.com/5G-MAG/rt-mbs-function) (MBSF), which drives it over Nmb8, and [rt-mbs-client](https://github.com/5G-MAG/rt-mbs-client) at the receiving end |
+| **Part of** | [5G Multicast Broadcast Services](https://www.5g-mag.com/reference-tools/5g-multicast-broadcast-services) |
+
 ## Introduction
 
-This repository provides a 5G MBS Transport Function which forms part of the MBS User Services. This NF provides the interfaces designated as Nmb2, Nmb8 and Nmb9 in the [3GPP TS 29.581 V18.5.0](https://www.3gpp.org/DynaReport/29581.htm) specification.
+This repository provides the MBS Transport Function. It takes content by PULL or PUSH, packages it into
+FLUTE object streams, and transmits them on the MBS session the MBSF has established, whether that
+session is broadcast or multicast.
+
+It is built on the [Open5GS](https://open5gs.org/) framework and registers with an NRF like any
+other network function.
+
+## Specification
+
+Built against these versions, named rather than referred to by release:
+
+- **3GPP TS 26.502 V18.6.0**, *5G multicast-broadcast services; User Service architecture*
+- **3GPP TS 29.581 V18.6.0**, *Nmb8 service API*
+- **3GPP TS 26.346 V18.2.0**, *MBMS protocols and codecs*, for the FLUTE and FDT profiling
 
-Additional information can be found at: https://5g-mag.github.io/Getting-Started/pages/5g-multicast-broadcast-services/
+Clause-by-clause coverage, and what is still absent, is recorded on the project page rather than
+here: <https://www.5g-mag.com/reference-tools/5g-multicast-broadcast-services>
 
 ## Install dependencies
 
@@ -24,6 +67,26 @@ sudo sh -c 'for i in cpp g++ gcc gcc-ar gcc-nm gcc-ranlib gcov gcov-dump gcov-to
 sudo python3 -m pip install --break-system-packages --upgrade meson
 ```
 
+### The build fetches the 5G APIs
+
+The OpenAPI bindings are generated at configure time from the 3GPP 5G APIs, which the build clones
+from `forge.3gpp.org`. The build therefore needs network access to that host, and Java, which is
+why `default-jdk` is in the list above.
+
+That host currently serves an **incomplete certificate chain**: it sends its own certificate but
+not the Sectigo intermediate that signs it. A browser fetches the missing intermediate by itself,
+but `git` and `curl` do not, so the clone fails with:
+
+```
+fatal: unable to access 'https://forge.3gpp.org/rep/all/5G_APIs.git/':
+  SSL certificate verification failed: certificate signer not trusted
+```
+
+If you see that, install the missing intermediate rather than disabling verification. On Debian
+and Ubuntu, fetch *Sectigo Public Server Authentication CA OV R36* from
+<https://crt.sh/>, put the PEM in `/usr/local/share/ca-certificates/` with a `.crt`
+extension, and run `sudo update-ca-certificates`.
+
 ## Downloading
 
 Release tar files can be downloaded from <https://github.com/5G-MAG/rt-mbs-transport-function/releases>.
@@ -33,30 +96,43 @@ The source can be obtained by cloning the github repository.
 For example to download the latest release you can use:
 
 ```bash
-cd ~
 git clone --recurse-submodules https://github.com/5G-MAG/rt-mbs-transport-function.git
+cd rt-mbs-transport-function
 ```
 
+`--recurse-submodules` is not optional: this repository carries `rt-common-shared` as a submodule
+and the build fails without it. If you have already cloned without it, run
+`git submodule update --init --recursive`.
+
+## Dependencies
+
+Two 5G-MAG libraries are pulled in by the build and fetched automatically. They are listed here
+because a version mismatch surfaces as a compile or link error rather than as a missing
+dependency.
+
+| Dependency | How | What it supplies |
+|---|---|---|
+| `rt-common-shared` | git submodule | the HTTP server and the shared Open5GS tooling, including the OpenAPI generator this build runs |
+| `rt-libflute` | meson wrap | the FLUTE transmitter, the TS 26.346 annex L.6 profiled FDT schema, the scheme-specific FEC OTI and the RFC 5053 Raptor scheme |
+
 ## Building
 
 The build process requires a working Internet connection as the API files are retrieved at build time.
 
 To build the 5G Data Collection Application Function from the source:
 
 ```bash
-cd ~/rt-mbs-transport-function
-meson build
+meson setup build
 ninja -C build
 ```
 
-**Note:** Errors during the `meson build` command are often caused by missing dependencies or a network issue while trying to retrieve the API files and `openapi-generator` JAR file. See the `~/rt-mbs-transport-function/build/meson-logs/meson-log.txt` log file for the errors in greater detail. Search for `generator-libspdc` to find the start of the API fetch sequence.
+**Note:** Errors during the `meson build` command are often caused by missing dependencies or a network issue while trying to retrieve the API files and `openapi-generator` JAR file. See the `build/meson-logs/meson-log.txt` log file for the errors in greater detail. Search for `generator-libspdc` to find the start of the API fetch sequence.
 
 ## Unit tests (optional)
 
 There are some unit tests that can be run using:
 
 ```bash
-cd ~/rt-mbs-transport-function
 meson test -C build --suite rt-mbs-transport-function
 ```
 
@@ -67,8 +143,7 @@ This will build the MBSTF (if not already built) and then will run the unit test
 To install the built MBS Transport Function as a system process:
 
 ```bash
-cd ~/rt-mbs-transport-function/build
-sudo meson install --no-rebuild
+sudo meson install -C build --no-rebuild
 ```
 
 ## Running
@@ -85,10 +160,26 @@ Make sure the IP address and port details of the NRF you are running are configu
 sudo /usr/local/bin/open5gs-mbstfd &
 ```
 
+## Configuration
+
+Configuration is a YAML file in the Open5GS style, installed as
+`/usr/local/etc/open5gs/mbstf.yaml` and passed with `-c` when running from a build tree. The
+sections that matter are `nrf`, which must point at a reachable NRF, and the MBSTF's own SBI
+address and the local address it sends FLUTE from.
+
 ## Development
 
 This project follows
 the [Gitflow workflow](https://www.atlassian.com/git/tutorials/comparing-workflows/gitflow-workflow). The
 `development` branch of this project serves as an integration branch for new features. Consequently, please make sure to
 switch to the `development` branch before starting the implementation of a new feature.
 
+## Contributing
+
+Contributions are welcome. How to raise an issue, fork the repository and open a pull request, and
+the Contributor License Agreement required before code can be merged, are described at
+<https://www.5g-mag.com/contributing>.
+
+## License
+
+See [LICENSE](LICENSE).

src/mbstf/Context.cc

@@ -48,6 +48,7 @@ Context::Context()
     ,servers()
     ,cacheControl({60, 60})
     ,totalMaxBitRateSoftLimit(100)
+    ,pathMtu(kDefaultPathMtu)
     ,consecutiveIngestFailuresBeforeDeactivate(5)
     ,packetModeSchedulingQueueSize(128*1024) // 128KB queue for rate smoothing
     ,manifestGlobals()
@@ -119,6 +120,18 @@ bool Context::parseConfig()
                     } else {
                         throw std::out_of_range("Bad configuration node at mbstf.totalMaxBitRateSoftLimit");
                     }
+                } else if (mbstf_key == "pathMtu") {
+                    Open5GSYamlIter mtu_iter(mbstf_iter);
+                    if (mtu_iter.type() == YAML_SCALAR_NODE) {
+                        std::string num_val(mtu_iter.value());
+                        size_t idx = 0;
+                        pathMtu = std::stoi(num_val, &idx);
+                        if (idx != num_val.size() || pathMtu <= 0) {
+                            throw std::out_of_range("Bad configuration value at mbstf.pathMtu");
+                        }
+                    } else {
+                        throw std::out_of_range("Bad configuration node at mbstf.pathMtu");
+                    }
                 } else if (mbstf_key == "consecutiveIngestFailuresBeforeDeactivate") {
                     Open5GSYamlIter failures_iter(mbstf_iter);
                     if (failures_iter.type() == YAML_SCALAR_NODE) {

src/mbstf/Context.hh

@@ -73,6 +73,37 @@ public:
         unsigned int defaultObjectMaxAge; // Use if not given by push/pull resource Cache-Control.
     } cacheControl;
     int totalMaxBitRateSoftLimit; //< total maximum bit rate this MBSTF ought to asked to handle
+    /**< MTU, in bytes, of the path a distribution session's packets travel to the receiver.
+     *
+     * FLUTE encoding symbols are sized from this, so a value larger than the path can carry puts
+     * every multi-symbol object into datagrams that do not arrive. RFC 5651 section 6.1: "However,
+     * network efficiency considerations recommend that the sender uses an as large as possible
+     * packet payload size, but in such a way that packets do not exceed the network's maximum
+     * transmission unit size (MTU), or when fragmentation coupled with packet loss might introduce
+     * severe inefficiency in the transmission."
+     *
+     * getsockopt(IP_MTU) on a socket to the session's tunnel address, or to its SSM destination
+     * when there is no tunnel, measures the first hop only. Where the MBSTF and the ingress point
+     * are co-located, which is every single-host deployment, that hop is the loopback interface
+     * and the answer is the loopback MTU, 65536 on Linux; and where a tunnel is configured the
+     * datagram is re-encapsulated and forwarded over a path the MBSTF cannot measure at all. The
+     * measurement is therefore a ceiling on the first hop, never a description of the whole path.
+     *
+     * kDefaultPathMtu is the default and the operator overrides it with mbstf.pathMtu. The
+     * discovered value is used in place of it only when it is smaller, since a first hop narrower
+     * than the stated path MTU is a real constraint while a wider one says nothing about the rest
+     * of the path.
+     */
+    int pathMtu;
+
+    /**< The path MTU assumed when the operator does not state one, in bytes.
+     *
+     * The conventional Ethernet MTU. No clause fixes it: it is a documented default, and a
+     * deployment whose path differs sets mbstf.pathMtu. Sizing symbols below the path MTU costs
+     * efficiency; sizing them above it costs delivery, so the default is the safe side of that.
+     */
+    static constexpr int kDefaultPathMtu = 1500;
+
     int consecutiveIngestFailuresBeforeDeactivate; //< The number of consecutive ingest failures allowed before the session aborts
     size_t packetModeSchedulingQueueSize; //< The maximum queue size for packet mode scheduling per DistSession
     struct {
@@ -82,6 +113,10 @@ public:
          */
         std::optional<std::chrono::milliseconds> manifestRepetitionRate = std::nullopt;
     } manifestGlobals; //< ManifestHandler global configuration (can be overridden by ManifestHandler implement specific config)
+    // TS 29.500 V18.10.0 cl.5.2.7.2/table 5.2.7.1-1: 413 (Payload Too Large) is mandatory for
+    // PATCH and POST. No clause, and no MBSTF documented default, names a byte limit
+    // -- unset means no limit is enforced, as before this option existed.
+    std::optional<size_t> maxRequestBodySize;
 
     /** Parse a configuration time duration string
      *

src/mbstf/DASHManifestHandler.cc

@@ -18,6 +18,7 @@
 #include <optional>
 #include <sstream>
 #include <stdexcept>
+#include <set>
 #include <string>
 #include <thread>
 #include <utility>
@@ -123,14 +124,46 @@ std::pair<ManifestHandler::time_type, ManifestHandler::ingest_list> DASHManifest
     static const std::string empty;
     auto current_time = std::chrono::system_clock::now();
     std::optional<std::chrono::system_clock::time_point> time_to_update;
-    std::string manifest_url;
+    /* The manifest's own URL, so the two comparisons below can recognise the MPD-refresh entry
+       addMPDRefreshToExtraPullObjects() adds to m_extraPullObjects from the same value. Left empty,
+       both comparisons test against "" and m_refreshMpd is never set, so a re-fetched MPD is
+       ingested without the handler being told its own manifest changed. */
+    const std::string manifest_url(m_manifest ? m_manifest->second.getFetchedUrl() : std::string());
     time_type fetch_time;
 
     std::list<SegmentEntry> media_segments;
     {
         std::lock_guard<std::recursive_mutex> guard(m_mpdMutex);
         media_segments = std::move(augmentSegmentAvailabilityList(m_mpd.selectedSegmentAvailability(), false, false, false));
     }
+    /* A live MPD lists a segment for the whole of its availability window, which tells a unicast client
+       it may still fetch that segment. It does not mean the segment still needs transmitting. Without
+       this filter the same segment is emitted on every pass until its window closes, and because
+       OBJECT_STREAMING removes an object from the ObjectStore once sent, findMetadataByURL() below
+       misses on each pass and builds a fresh object, which the packager then sends under a fresh TOI.
+       A receiver cannot combine symbols across TOIs, so those copies are not redundancy: each is a
+       separate incomplete object, and together they crowd out the bandwidth the first copy needed.
+       Measured before this filter: 4.0 TOIs per object on average and up to 15. */
+    pruneSentSegments(media_segments);
+    media_segments.remove_if([this](const SegmentEntry &seg) {
+            try {
+                return m_sentSegmentUrls.find(seg.segmentURL()) != m_sentSegmentUrls.end();
+            } catch (std::domain_error&) {
+                return false; /* leave a malformed URL to the existing handling further down */
+            }
+        });
+
+    /* Only segments the MPD itself advertises are suppressed on a later pass. The entries appended
+       from m_extraPullObjects below -- the MPD refresh and the initialisation segments -- are meant to
+       repeat, and are deliberately not recorded. */
+    std::set<std::string> media_segment_urls;
+    for (const auto &seg : media_segments) {
+        try {
+            media_segment_urls.insert(seg.segmentURL());
+        } catch (std::domain_error&) {
+        }
+    }
+
     media_segments.insert(media_segments.end(), m_extraPullObjects.begin(), m_extraPullObjects.end());
     for (auto &ms: media_segments) {
         if(ms.availabilityStartTime() < current_time)
@@ -171,6 +204,7 @@ std::pair<ManifestHandler::time_type, ManifestHandler::ingest_list> DASHManifest
                                       obj_dist_base_url, first_media_segment.availabilityEndTime(), first_media_segment.forceRecache(), first_media_segment.keepAfterSend(), first_media_segment.compressEntry());
         }
         removeExtraPullObjectsEntry(first_media_segment);
+        if (media_segment_urls.find(segment_url) != media_segment_urls.end()) m_sentSegmentUrls.insert(segment_url);
 
         try {
             if (first_media_segment.segmentURL() == manifest_url) m_refreshMpd = true;
@@ -185,6 +219,7 @@ std::pair<ManifestHandler::time_type, ManifestHandler::ingest_list> DASHManifest
             segment_url = it->segmentURL();
             existing_obj = object_store->findMetadataByURL(segment_url);
             removeExtraPullObjectsEntry(*it);
+            if (media_segment_urls.find(segment_url) != media_segment_urls.end()) m_sentSegmentUrls.insert(segment_url);
             if (existing_obj) {
                 ingest_items.emplace_back(*existing_obj, it->availabilityEndTime(), it->forceRecache(), it->keepAfterSend(), it->compressEntry());
             } else {
@@ -198,6 +233,28 @@ std::pair<ManifestHandler::time_type, ManifestHandler::ingest_list> DASHManifest
     return std::make_pair(fetch_time, ingest_items);
 }
 
+void DASHManifestHandler::pruneSentSegments(const std::list<SegmentEntry> &current_segments)
+{
+    /* Bounds m_sentSegmentUrls by the manifest itself: a segment the MPD no longer advertises can no
+       longer be re-emitted by the filter above, so remembering it serves nothing. Without this the set
+       grows for the lifetime of a live session. */
+    if (m_sentSegmentUrls.empty()) return;
+    std::set<std::string> still_listed;
+    for (const auto &seg : current_segments) {
+        try {
+            still_listed.insert(seg.segmentURL());
+        } catch (std::domain_error&) {
+        }
+    }
+    for (auto it = m_sentSegmentUrls.begin(); it != m_sentSegmentUrls.end();) {
+        if (still_listed.find(*it) == still_listed.end()) {
+            it = m_sentSegmentUrls.erase(it);
+        } else {
+            ++it;
+        }
+    }
+}
+
 void DASHManifestHandler::addMPDRefreshToExtraPullObjects()
 {
     std::lock_guard<std::recursive_mutex> guard(m_mpdMutex);

src/mbstf/DASHManifestHandler.hh

@@ -12,6 +12,7 @@
  * https://drive.google.com/file/d/1cinCiA778IErENZ3JN52VFW-1ffHpx7Z/view
  */
 #include <list>
+#include <set>
 #include <string>
 #include <thread>
 #include <utility>
@@ -95,6 +96,25 @@ private:
     bool m_refreshMpd;
     ManifestHandler::time_type m_mpdReceivedTime;
     std::list<SegmentEntry> m_extraPullObjects;
+    /** Media segment URLs already handed to the ingester in this session.
+     *
+     * A live MPD keeps a segment listed for the whole of its availability window, which is what tells
+     * a unicast client it may still fetch it. It does not mean the segment still needs sending. In
+     * OBJECT_STREAMING an object is removed from the ObjectStore once it has been sent
+     * (ObjectStreamingController leaves Metadata::keepAfterSend() at false, unlike the carousel), so
+     * the store cannot answer "has this been sent already?" either: findMetadataByURL() misses, a
+     * second ObjectStore object is created for the same URL, and it goes out under a second TOI.
+     *
+     * A receiver cannot combine symbols across TOIs. RFC 3926 clause 3.1: "Note that each object is
+     * associated with a unique TOI within the scope of a session." Each copy is therefore a separate,
+     * independently incomplete object rather than redundancy, and the copies consume the bearer that
+     * the first copy needed.
+     *
+     * Bounded by the MPD's own availability window: pruneSentSegments() drops every entry the current
+     * manifest no longer lists, so this holds at most one string per segment currently advertised.
+     */
+    std::set<std::string> m_sentSegmentUrls;
+    void pruneSentSegments(const std::list<SegmentEntry> &current_segments);
 };
 
 MBSTF_NAMESPACE_STOP