MT#55283 add rollback NG message

daemon/call.c

@@ -69,7 +69,6 @@ static int64_t add_ongoing_calls_dur_in_interval(int64_t interval_start, int64_t
 static void __call_free(call_t *p);
 static void __call_cleanup(call_t *c);
 static void __monologue_stop(struct call_monologue *ml);
-static void media_stop(struct call_media *m);
 __attribute__((nonnull(1, 2, 4)))
 static struct media_subscription *__subscribe_medias_both_ways(struct call_media * a, struct call_media * b,
 		bool is_offer, medias_q *);
@@ -5182,7 +5181,7 @@ static void __call_cleanup(call_t *c) {
 	for (__auto_type l = c->medias.head; l; l = l->next) {
 		struct call_media *md = l->data;
 		ice_shutdown(&md->ice_agent);
-		media_stop(md);
+		call_media_stop(md);
 		t38_gateway_put(&md->t38_gateway);
 		audio_player_free(md);
 		mutex_destroy(&md->dtmf_lock);
@@ -5498,6 +5497,7 @@ static void __call_free(call_t *c) {
 
 	//ilog(LOG_DEBUG, "freeing main call struct");
 
+	call_checkpoint_free_all(c);
 	obj_release(c->dtls_cert);
 	mqtt_timer_stop(&c->mqtt_timer);
 
@@ -5748,6 +5748,9 @@ static bool call_merge(call_t *call, call_t *call2) {
 	// move buffers
 	bencode_buffer_merge(&call->buffer, &call2->buffer);
 
+	// the ids below are about to be renumbered, and a snapshot is keyed on them
+	call_checkpoint_free_all(call2);
+
 	// move all contained objects: we have to renumber all unique IDs, and redirect any
 	// `call` pointers
 
@@ -6418,7 +6421,7 @@ int call_get_mono_dialogue(struct call_monologue *monologues[2],
 	return call_get_dialogue(monologues, call, callid, fromtag, totag, viabranch, flags, ep);
 }
 
-static void media_stop(struct call_media *m) {
+void call_media_stop(struct call_media *m) {
 	if (!m)
 		return;
 	t38_gateway_stop(m->t38_gateway);
@@ -6443,7 +6446,7 @@ static void monologue_stop(struct call_monologue *ml, bool stop_media_subscriber
 	__monologue_stop(ml);
 	for (unsigned int i = 0; i < ml->medias->len; i++)
 	{
-		media_stop(ml->medias->pdata[i]);
+		call_media_stop(ml->medias->pdata[i]);
 	}
 	/* monologue's subscribers */
 	if (stop_media_subscribers) {
@@ -6453,7 +6456,7 @@ static void monologue_stop(struct call_monologue *ml, bool stop_media_subscriber
 			if (!media)
 				continue;
 			IQUEUE_FOREACH(&media->media_subscribers, ms) {
-				media_stop(ms->media);
+				call_media_stop(ms->media);
 				__monologue_stop(ms->monologue);
 			}
 		}
@@ -6869,3 +6872,66 @@ void call_q_unlock_release(call_q *calls) {
 		call_unlock_release(call);
 	}
 }
+
+
+static void checkpoint_clear_snapshot(struct call_checkpoint *cp) {
+	redis_snapshot_free(&cp->snapshot);
+	cp->pending = false;
+}
+
+static void checkpoint_offer_one(call_t *call, struct call_monologue *ml, bool enable) {
+	if (!ml)
+		return;
+	if (!ml->checkpoint) {
+		if (!enable)
+			return;
+		ml->checkpoint = g_new0(__typeof(*ml->checkpoint), 1);
+	}
+	// consecutive offers belong to the same uncommitted exchange: keep the
+	// committed snapshot, or a later rollback restores a rejected offer
+	if (ml->checkpoint->pending)
+		return;
+	checkpoint_clear_snapshot(ml->checkpoint);
+	ml->checkpoint->snapshot = redis_snapshot_encode(call, ml);
+	ml->checkpoint->pending = true;
+}
+
+void call_checkpoint_offer(call_t *call, struct call_monologue *offerer,
+		struct call_monologue *answerer, bool enable)
+{
+	// a dialogue is tracked if either side is, so both are checkpointed together
+	bool tracked = enable
+			|| (offerer && offerer->checkpoint)
+			|| (answerer && answerer->checkpoint);
+	checkpoint_offer_one(call, offerer, tracked);
+	checkpoint_offer_one(call, answerer, tracked);
+}
+
+static void checkpoint_commit_one(struct call_monologue *ml) {
+	if (ml && ml->checkpoint && ml->checkpoint->pending)
+		checkpoint_clear_snapshot(ml->checkpoint);
+}
+
+void call_checkpoint_answer(call_t *call, struct call_monologue *a, struct call_monologue *b) {
+	checkpoint_commit_one(a);
+	checkpoint_commit_one(b);
+}
+
+int call_checkpoint_rollback(call_t *call, struct call_monologue *a, struct call_monologue *b) {
+	if (!redis_snapshot_apply(call, a, b))
+		return 0;
+
+	call->last_signal_us = rtpe_now;
+	return 1;
+}
+
+void call_checkpoint_free_all(call_t *call) {
+	for (__auto_type l = call->monologues.head; l; l = l->next) {
+		struct call_monologue *ml = l->data;
+		if (!ml->checkpoint)
+			continue;
+		redis_snapshot_free(&ml->checkpoint->snapshot);
+		g_free(ml->checkpoint);
+		ml->checkpoint = NULL;
+	}
+}

daemon/call_flags.c

@@ -526,6 +526,8 @@ static const char *call_ng_flags_supports(str *s, unsigned int idx, helper_arg a
 	sdp_ng_flags *out = arg.flags;
 	if (!str_cmp(s, "load limit"))
 		out->supports_load_limit = true;
+	else if (!str_cmp(s, "rollback"))
+		out->supports_rollback = true;
 	else
 		ilog(LOG_INFO | LOG_FLAG_LIMIT, "Optional feature '" STR_FORMAT "' not supported",
 				STR_FMT(s));
@@ -909,6 +911,10 @@ const char *call_ng_flags_flags(str *s, unsigned int idx, helper_arg arg) {
 		case CSH_LOOKUP("reset"):
 			out->reset = true;
 			break;
+		case CSH_LOOKUP("track-state"):
+		case CSH_LOOKUP("track state"):
+			out->track_state = true;
+			break;
 		case CSH_LOOKUP("single-codec"):
 		case CSH_LOOKUP("single codec"):
 			out->single_codec = true;

daemon/call_interfaces.c

@@ -652,10 +652,18 @@ static const char *call_offer_answer_ng(ng_command_ctx_t *ctx, const char *addr)
 		t_hash_table_insert(call->endpoints, memory_arena_objdup(streams.head->data->rtp_endpoint),
 				from_ml);
 
+	if (flags.opmode == OP_OFFER) {
+		// opened before SDP processing, to capture the pre-offer state. left
+		// pending if the offer is rejected: it's still the committed state
+		call_checkpoint_offer(call, from_ml, to_ml, flags.track_state);
+	}
+
 	struct recording *recording = NULL;
 
 	/* offer/answer model processing */
 	if ((ret = monologue_offer_answer(monologues, &streams, &flags)) == 0) {
+		if (flags.opmode == OP_ANSWER)
+			call_checkpoint_answer(call, from_ml, to_ml);
 		update_metadata_monologue(from_ml, &flags);
 		detect_setup_recording(call, &flags);
 
@@ -681,6 +689,10 @@ static const char *call_offer_answer_ng(ng_command_ctx_t *ctx, const char *addr)
 		/* place return output SDP */
 		ctx->ngbuf->sdp_out = sdp_out.s;
 		ctx->parser_ctx.parser->dict_add_str(output, "sdp", &sdp_out);
+		if (flags.supports_rollback) {
+			parser_arg supported = parser->dict_add_list(output, "supported");
+			parser->list_add_string(supported, "rollback");
+		}
 
 		meta_write_sdp_after(recording, &sdp_out, from_ml, flags.opmode);
 
@@ -734,6 +746,59 @@ const char *call_answer_ng(ng_command_ctx_t *ctx) {
 	return call_offer_answer_ng(ctx, NULL);
 }
 
+static bool monologue_has_tag(const struct call_monologue *ml, const str *tag) {
+	if (!str_cmp_str(&ml->tag, tag))
+		return true;
+	for (__auto_type l = ml->tag_aliases.head; l; l = l->next) {
+		if (!str_cmp_str(l->data, tag))
+			return true;
+	}
+	return false;
+}
+
+const char *call_rollback_ng(ng_command_ctx_t *ctx) {
+	const ng_parser_t *parser = ctx->parser_ctx.parser;
+	str call_id = parser->dict_get_str(ctx->req, "call-id");
+	str from_tag = parser->dict_get_str(ctx->req, "from-tag");
+	str to_tag = parser->dict_get_str(ctx->req, "to-tag");
+	str via_branch = parser->dict_get_str(ctx->req, "via-branch");
+
+	if (!call_id.len)
+		return "No call-id in message";
+	if (!from_tag.len)
+		return "No from-tag in message";
+	if (!to_tag.len)
+		return "No to-tag in message";
+
+	call_t *call = call_get(&call_id);
+	if (!call)
+		return "Unknown call-id";
+
+	struct call_monologue *from_ml = call_get_monologue(call, &from_tag);
+	struct call_monologue *to_ml = via_branch.len
+		? t_hash_table_lookup(call->viabranches, &via_branch)
+		: call_get_monologue(call, &to_tag);
+	// call_get_monologue() is keyed on the tag, so from_ml carries it by
+	// construction. to_ml may have come from the viabranch table instead.
+	if (!from_ml || !to_ml || from_ml == to_ml
+			|| !monologue_has_tag(to_ml, &to_tag)
+			|| !g_hash_table_contains(from_ml->associated_tags, to_ml))
+	{
+		rwlock_unlock_w(&call->master_lock);
+		obj_put(call);
+		return "Unknown dialogue";
+	}
+	if (rtpe_config.active_switchover && IS_FOREIGN_CALL(call))
+		call_make_own_foreign(call, false);
+
+	int rolled_back = call_checkpoint_rollback(call, from_ml, to_ml);
+	parser->dict_add_int(ctx->resp, "rolled-back", rolled_back);
+	rwlock_unlock_w(&call->master_lock);
+	redis_update_onekey(call, rtpe_redis_write);
+	obj_put(call);
+	return NULL;
+}
+
 const char *call_delete_ng(ng_command_ctx_t *ctx) {
 	g_auto(sdp_ng_flags) rtpp_flags;
 	parser_arg output = ctx->resp;

daemon/ice.c

@@ -53,6 +53,7 @@ static void __agent_schedule(struct ice_agent *ag, int64_t);
 static void __agent_schedule_abs(struct ice_agent *ag, int64_t tv);
 static void __agent_deschedule(struct ice_agent *ag);
 static void __ice_agent_free_components(struct ice_agent *ag);
+static void __ice_pairings(struct ice_agent *ag);
 static void __agent_shutdown(struct ice_agent *ag);
 static void ice_agents_timer_run(void *);
 
@@ -358,7 +359,7 @@ TYPED_GHASHTABLE_IMPL(foundation_ht, __found_hash, __found_equal, NULL, NULL)
 TYPED_GHASHTABLE_IMPL(priority_ht, g_direct_hash, g_direct_equal, NULL, NULL)
 TYPED_GHASHTABLE_IMPL(transaction_ht, __trans_hash, __trans_equal, NULL, NULL)
 
-static void __ice_agent_initialize(struct ice_agent *ag) {
+static void __ice_agent_initialize(struct ice_agent *ag, bool generate_credentials) {
 	struct call_media *media = ag->media;
 	call_t *call = ag->call;
 
@@ -377,8 +378,10 @@ static void __ice_agent_initialize(struct ice_agent *ag) {
 	ag->succeeded_pairs = g_tree_new(__pair_prio_cmp);
 	ag->all_pairs = g_tree_new(__pair_prio_cmp);
 
-	create_random_ice_string(call, &ag->ufrag[1], 8);
-	create_random_ice_string(call, &ag->pwd[1], 26);
+	if (generate_credentials) {
+		create_random_ice_string(call, &ag->ufrag[1], 8);
+		create_random_ice_string(call, &ag->pwd[1], 26);
+	}
 
 	atomic64_set_na(&ag->last_activity, rtpe_now);
 }
@@ -394,7 +397,7 @@ static struct ice_agent *__ice_agent_new(struct call_media *media) {
 	ag->media = media;
 	mutex_init(&ag->lock);
 
-	__ice_agent_initialize(ag);
+	__ice_agent_initialize(ag, true);
 
 	return ag;
 }
@@ -416,14 +419,40 @@ static unsigned int __copy_cand(call_t *call, struct ice_candidate *dst, const s
 	return eq ? 0 : 1;
 }
 
-static void __ice_reset(struct ice_agent *ag) {
+// generate_credentials = new local credentials; not wanted for a rollback,
+// which installs the ones the committed exchange agreed
+static void __ice_reset(struct ice_agent *ag, bool generate_credentials) {
 	__agent_deschedule(ag);
 	AGENT_CLEAR3(ag, COMPLETED, NOMINATING, USABLE);
 	__ice_agent_free_components(ag);
 	ZERO(ag->active_components);
 	ag->start_nominating = 0;
 	ag->tt_obj.last_run = 0;
-	__ice_agent_initialize(ag);
+	__ice_agent_initialize(ag, generate_credentials);
+}
+
+/* called with the call lock held in W, hence agent doesn't need to be locked */
+void ice_rollback(struct ice_agent *ag, const str ufrag[2], const str pwd[2],
+		const candidate_q *candidates)
+{
+	if (!ag)
+		return;
+
+	__ice_reset(ag, false);
+	memcpy(ag->ufrag, ufrag, sizeof(ag->ufrag));
+	memcpy(ag->pwd, pwd, sizeof(ag->pwd));
+
+	for (__auto_type l = candidates->head; l; l = l->next) {
+		struct ice_candidate *copy = g_new(__typeof(*copy), 1);
+		*copy = *(struct ice_candidate *) l->data;
+		t_hash_table_insert(ag->candidate_hash, copy, copy);
+		t_hash_table_insert(ag->cand_prio_hash, GUINT_TO_POINTER(copy->priority), copy);
+		t_hash_table_insert(ag->foundation_hash, copy, copy);
+		t_queue_push_tail(&ag->remote_candidates, copy);
+		ag->active_components = MAX(ag->active_components, copy->component_id);
+	}
+	__ice_pairings(ag);
+	ice_start(ag);
 }
 
 /* if the other side did a restart */
@@ -434,7 +463,7 @@ static void __ice_restart(struct ice_agent *ag) {
 	ag->pwd[0] = STR_NULL;
 	ag->ufrag[1] = STR_NULL;
 	ag->pwd[1] = STR_NULL;
-	__ice_reset(ag);
+	__ice_reset(ag, true);
 }
 
 /* if we're doing a restart */
@@ -443,7 +472,7 @@ void ice_restart(struct ice_agent *ag) {
 
 	ag->ufrag[1] = STR_NULL;
 	ag->pwd[1] = STR_NULL;
-	__ice_reset(ag);
+	__ice_reset(ag, true);
 }
 
 /* called with the call lock held in W, hence agent doesn't need to be locked */