audioaggregator: Resync on the next buffer when dropping a buffer on discont resyncing

If a buffer is dropped during resyncing on a discont because either its
end offset is already before the current output offset of the
aggregator or because it fully overlaps with the part of the current
output buffer that was already filled, then don't just assume that the
next buffer is going to start at exactly the expected offset. It might
still require some more dropping of samples.

This caused the input to be mixed with an offset to its actual position
in the output stream, causing additional latency and wrong
synchronization between the different input streams.

Instead consider each buffer after a discont as a discont until the
aggregator actually resynced and starts mixing samples from the input
again.

Also update the start output offset of a new input buffer if samples
have to be dropped at the beginning. Otherwise it might be mixed too
early into the output and overwrite part of the output buffer that
already took samples from this input into account.

Fixes https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/-/issues/912
which is a regression introduced by https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/-/merge_requests/1180/

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/-/merge_requests/1228>

2 files changed, 96 insertions, 19 deletions

diff --git a/gst-libs/gst/audio/gstaudioaggregator.c b/gst-libs/gst/audio/gstaudioaggregator.c
index 2668c51b1..192caa3b9 100644
--- a/gst-libs/gst/audio/gstaudioaggregator.c
+++ b/gst-libs/gst/audio/gstaudioaggregator.c
@@ -105,7 +105,8 @@ struct _GstAudioAggregatorPadPrivate
                                    of this input_buffer would be placed. */
 
   guint64 next_offset;          /* Next expected sample offset relative to
-                                   pad.segment.start */
+                                   pad.segment.start. This is -1 when resyncing is
+                                   needed, e.g. because of a previous discont. */
 
   /* Last time we noticed a discont */
   GstClockTime discont_time;
@@ -1669,7 +1670,7 @@ gst_audio_aggregator_fill_buffer (GstAudioAggregator * aagg,
       GST_DEBUG_OBJECT (pad, "Have discont. Expected %"
           G_GUINT64_FORMAT ", got %" G_GUINT64_FORMAT,
           pad->priv->next_offset, start_offset);
-    pad->priv->next_offset = end_offset;
+    pad->priv->next_offset = -1;
   } else {
     pad->priv->next_offset += pad->priv->size;
   }
@@ -1750,6 +1751,8 @@ gst_audio_aggregator_fill_buffer (GstAudioAggregator * aagg,
       }
 
       pad->priv->position += diff;
+      if (start_output_offset != -1)
+        start_output_offset += diff;
       if (pad->priv->position >= pad->priv->size) {
         /* Empty buffer, drop */
         pad->priv->position = 0;
@@ -1761,14 +1764,14 @@ gst_audio_aggregator_fill_buffer (GstAudioAggregator * aagg,
       }
     }
 
-    if (start_output_offset == -1 || start_output_offset < aagg->priv->offset)
+    if (start_output_offset == -1)
       pad->priv->output_offset = aagg->priv->offset;
-    else if (pad->priv->output_offset != -1)
-      pad->priv->output_offset = MAX (pad->priv->output_offset,
-          start_output_offset);
     else
       pad->priv->output_offset = start_output_offset;
 
+    if (pad->priv->next_offset == -1)
+      pad->priv->next_offset = end_offset;
+
     GST_DEBUG_OBJECT (pad,
         "Buffer resynced: Pad offset %" G_GUINT64_FORMAT
         ", current audio aggregator offset %" G_GINT64_FORMAT,
diff --git a/tests/check/elements/audiomixer.c b/tests/check/elements/audiomixer.c
index 62fc1c39d..c64947ac5 100644
--- a/tests/check/elements/audiomixer.c
+++ b/tests/check/elements/audiomixer.c
@@ -1459,7 +1459,7 @@ send_buffers_sync_discont_backwards (GstPad * pad1, GstPad * pad2)
   GstBuffer *buffer;
   GstFlowReturn ret;
 
-  buffer = new_buffer (2300, 1, 1 * GST_SECOND, 1.25 * GST_SECOND, 0);
+  buffer = new_buffer (2300, 1, 1 * GST_SECOND, 1.15 * GST_SECOND, 0);
   ret = gst_pad_chain (pad1, buffer);
   ck_assert_int_eq (ret, GST_FLOW_OK);
 
@@ -1494,23 +1494,23 @@ check_buffers_sync_discont_backwards (GList * received_buffers)
     gst_buffer_map (buffer, &map, GST_MAP_READ);
 
     if (i == 0 && GST_BUFFER_TIMESTAMP (buffer) == 0) {
-      fail_unless (map.data[0] == 0);
-      fail_unless (map.data[map.size - 1] == 0);
+      fail_unless_equals_int (map.data[0], 0);
+      fail_unless_equals_int (map.data[map.size - 1], 0);
     } else if (i == 1 && GST_BUFFER_TIMESTAMP (buffer) == 500 * GST_MSECOND) {
-      fail_unless (map.data[0] == 0);
-      fail_unless (map.data[map.size - 1] == 0);
+      fail_unless_equals_int (map.data[0], 0);
+      fail_unless_equals_int (map.data[map.size - 1], 0);
     } else if (i == 2 && GST_BUFFER_TIMESTAMP (buffer) == 1000 * GST_MSECOND) {
-      fail_unless (map.data[0] == 1);
-      fail_unless (map.data[map.size - 1] == 1);
+      fail_unless_equals_int (map.data[0], 1);
+      fail_unless_equals_int (map.data[map.size - 1], 1);
     } else if (i == 3 && GST_BUFFER_TIMESTAMP (buffer) == 1500 * GST_MSECOND) {
-      fail_unless (map.data[0] == 1);
-      fail_unless (map.data[map.size - 1] == 1);
+      fail_unless_equals_int (map.data[0], 1);
+      fail_unless_equals_int (map.data[map.size - 1], 1);
     } else if (i == 4 && GST_BUFFER_TIMESTAMP (buffer) == 2000 * GST_MSECOND) {
-      fail_unless (map.data[0] == 2);
-      fail_unless (map.data[map.size - 1] == 2);
+      fail_unless_equals_int (map.data[0], 2);
+      fail_unless_equals_int (map.data[map.size - 1], 2);
     } else if (i == 5 && GST_BUFFER_TIMESTAMP (buffer) == 2500 * GST_MSECOND) {
-      fail_unless (map.data[0] == 2);
-      fail_unless (map.data[map.size - 1] == 2);
+      fail_unless_equals_int (map.data[0], 2);
+      fail_unless_equals_int (map.data[map.size - 1], 2);
     } else {
       g_assert_not_reached ();
     }
@@ -1528,6 +1528,78 @@ GST_START_TEST (test_sync_discont_backwards)
 
 GST_END_TEST;
 
+static void
+send_buffers_sync_discont_and_drop_backwards (GstPad * pad1, GstPad * pad2)
+{
+  GstBuffer *buffer;
+  GstFlowReturn ret;
+
+  buffer = new_buffer (2500, 1, 1 * GST_SECOND, 1.25 * GST_SECOND, 0);
+  ret = gst_pad_chain (pad1, buffer);
+  ck_assert_int_eq (ret, GST_FLOW_OK);
+
+  buffer = new_buffer (400, 1, 2 * GST_SECOND, 0.2 * GST_SECOND,
+      GST_BUFFER_FLAG_DISCONT);
+  ret = gst_pad_chain (pad1, buffer);
+  ck_assert_int_eq (ret, GST_FLOW_OK);
+
+  buffer = new_buffer (1600, 1, 2.2 * GST_SECOND, 0.8 * GST_SECOND, 0);
+  ret = gst_pad_chain (pad1, buffer);
+  ck_assert_int_eq (ret, GST_FLOW_OK);
+
+  gst_pad_send_event (pad1, gst_event_new_eos ());
+
+  buffer = new_buffer (2000, 1, 2 * GST_SECOND, 1 * GST_SECOND, 0);
+  ret = gst_pad_chain (pad2, buffer);
+  ck_assert_int_eq (ret, GST_FLOW_OK);
+
+  gst_pad_send_event (pad2, gst_event_new_eos ());
+}
+
+GST_START_TEST (test_sync_discont_and_drop_backwards)
+{
+  run_sync_test (send_buffers_sync_discont_and_drop_backwards,
+      check_buffers_sync_discont_backwards);
+}
+
+GST_END_TEST;
+
+static void
+send_buffers_sync_discont_and_drop_before_output_backwards (GstPad * pad1,
+    GstPad * pad2)
+{
+  GstBuffer *buffer;
+  GstFlowReturn ret;
+
+  buffer = new_buffer (2500, 1, 1 * GST_SECOND, 1.25 * GST_SECOND, 0);
+  ret = gst_pad_chain (pad1, buffer);
+  ck_assert_int_eq (ret, GST_FLOW_OK);
+
+  buffer = new_buffer (800, 1, 1.5 * GST_SECOND, 0.4 * GST_SECOND,
+      GST_BUFFER_FLAG_DISCONT);
+  ret = gst_pad_chain (pad1, buffer);
+  ck_assert_int_eq (ret, GST_FLOW_OK);
+
+  buffer = new_buffer (2200, 1, 1.9 * GST_SECOND, 1.1 * GST_SECOND, 0);
+  ret = gst_pad_chain (pad1, buffer);
+  ck_assert_int_eq (ret, GST_FLOW_OK);
+
+  gst_pad_send_event (pad1, gst_event_new_eos ());
+
+  buffer = new_buffer (2000, 1, 2 * GST_SECOND, 1 * GST_SECOND, 0);
+  ret = gst_pad_chain (pad2, buffer);
+  ck_assert_int_eq (ret, GST_FLOW_OK);
+
+  gst_pad_send_event (pad2, gst_event_new_eos ());
+}
+
+GST_START_TEST (test_sync_discont_and_drop_before_output_backwards)
+{
+  run_sync_test (send_buffers_sync_discont_and_drop_before_output_backwards,
+      check_buffers_sync_discont_backwards);
+}
+
+GST_END_TEST;
 
 static void
 send_buffers_sync_unaligned (GstPad * pad1, GstPad * pad2)
@@ -2047,6 +2119,8 @@ audiomixer_suite (void)
   tcase_add_test (tc_chain, test_sync);
   tcase_add_test (tc_chain, test_sync_discont);
   tcase_add_test (tc_chain, test_sync_discont_backwards);
+  tcase_add_test (tc_chain, test_sync_discont_and_drop_backwards);
+  tcase_add_test (tc_chain, test_sync_discont_and_drop_before_output_backwards);
   tcase_add_test (tc_chain, test_sync_unaligned);
   tcase_add_test (tc_chain, test_segment_base_handling);
   tcase_add_test (tc_chain, test_sinkpad_property_controller);