diff --git a/daemon/call.c b/daemon/call.c index fc8188d24..a7b40f875 100644 --- a/daemon/call.c +++ b/daemon/call.c @@ -6871,67 +6871,60 @@ void call_q_unlock_release(call_q *calls) { } -static struct call_checkpoint *checkpoint_find(call_t *call, struct call_monologue *a, - struct call_monologue *b) -{ - for (struct call_checkpoint *cp = call->checkpoints; cp; cp = cp->next) { - if ((cp->offerer == a && cp->answerer == b) || (cp->offerer == b && cp->answerer == a)) - return cp; - } - return NULL; -} - static void checkpoint_clear_snapshot(struct call_checkpoint *cp) { redis_snapshot_free(&cp->snapshot); cp->pending = false; } -void call_checkpoint_offer(call_t *call, struct call_monologue *offerer, - struct call_monologue *answerer, bool enable) -{ - struct call_checkpoint *cp = checkpoint_find(call, offerer, answerer); - if (!cp && !enable) +static void checkpoint_offer_one(call_t *call, struct call_monologue *ml, bool enable) { + if (!ml) return; - if (!cp) { - cp = g_new0(__typeof(*cp), 1); - cp->next = call->checkpoints; - call->checkpoints = cp; + 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 (cp->pending) + if (ml->checkpoint->pending) return; - checkpoint_clear_snapshot(cp); - cp->offerer = offerer; - cp->answerer = answerer; - cp->snapshot = redis_snapshot_encode(call, offerer, answerer); - cp->pending = true; + 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) +{ + checkpoint_offer_one(call, offerer, enable); + checkpoint_offer_one(call, answerer, enable); +} + +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) { - struct call_checkpoint *cp = checkpoint_find(call, a, b); - if (cp && cp->pending) - checkpoint_clear_snapshot(cp); + checkpoint_commit_one(a); + checkpoint_commit_one(b); } int call_checkpoint_rollback(call_t *call, struct call_monologue *a, struct call_monologue *b) { - struct call_checkpoint *cp = checkpoint_find(call, a, b); - if (!cp || !cp->pending) - return 0; - - if (!redis_snapshot_apply(call, &cp->snapshot, cp->offerer, cp->answerer)) + if (!redis_snapshot_apply(call, a, b)) return 0; - checkpoint_clear_snapshot(cp); call->last_signal_us = rtpe_now; return 1; } void call_checkpoint_free_all(call_t *call) { - while (call->checkpoints) { - struct call_checkpoint *cp = call->checkpoints; - call->checkpoints = cp->next; - redis_snapshot_free(&cp->snapshot); - g_free(cp); + 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; } } diff --git a/daemon/redis.c b/daemon/redis.c index 3aba11cb9..3141361a8 100644 --- a/daemon/redis.c +++ b/daemon/redis.c @@ -2147,23 +2147,17 @@ static int checkpoint_get_int(int64_t *out, const struct redis_hash *h, const ch return 0; } -static int redis_restore_checkpoints(call_t *c, const struct redis_hash *call, - parser_arg root) -{ - int64_t num = 0; - if (checkpoint_get_int(&num, call, "num_checkpoints")) - return 0; /* written by a version that had none: not an error */ - - for (int64_t i = 0; i < num; i++) { +static int redis_restore_checkpoints(call_t *c, parser_arg root) { + for (__auto_type l = c->monologues.head; l; l = l->next) { + struct call_monologue *ml = l->data; struct redis_hash rh; - if (json_get_hash(&rh, "checkpoint", (unsigned int) i, root)) - return -1; + // absent for a call written by a version that had no checkpoints + if (json_get_hash(&rh, "checkpoint", ml->unique_id, root)) + continue; - int64_t offerer = -1, answerer = -1, pending = 0; + int64_t pending = 0; str snap = STR_NULL; - int bad = checkpoint_get_int(&offerer, &rh, "offerer") - || checkpoint_get_int(&answerer, &rh, "answerer") - || checkpoint_get_int(&pending, &rh, "pending"); + int bad = checkpoint_get_int(&pending, &rh, "pending"); /* the hash owns its values; copy out before it's destroyed */ if (!bad) { str stored; @@ -2176,30 +2170,12 @@ static int redis_restore_checkpoints(call_t *c, const struct redis_hash *call, return -1; } - struct call_monologue *a = NULL, *b = NULL; - for (__auto_type l = c->monologues.head; l; l = l->next) { - struct call_monologue *ml = l->data; - if (ml->unique_id == (unsigned int) offerer) - a = ml; - else if (ml->unique_id == (unsigned int) answerer) - b = ml; - } - if (!a || !b) { - str_free_dup(&snap); - return -1; - } - - - struct call_checkpoint *cp = g_new0(__typeof(*cp), 1); - cp->offerer = a; - cp->answerer = b; - cp->pending = pending && snap.len; - if (cp->pending) - cp->snapshot = snap; + ml->checkpoint = g_new0(__typeof(*ml->checkpoint), 1); + ml->checkpoint->pending = pending && snap.len; + if (ml->checkpoint->pending) + ml->checkpoint->snapshot = snap; else str_free_dup(&snap); - cp->next = c->checkpoints; - c->checkpoints = cp; } return 0; } @@ -2208,6 +2184,7 @@ struct redis_parsed_record { JsonParser *json; bencode_buffer_t benc; bool benc_valid; + const ng_parser_t *parser; }; static const char *redis_parse_record(const str *record, parser_arg *root, @@ -2225,7 +2202,7 @@ static const char *redis_parse_record(const str *record, parser_arg *root, if (!json_root) return "could not read JSON data"; root->json = json_root; - redis_parser = &ng_parser_json; + redis_parser = out->parser = &ng_parser_json; return NULL; } @@ -2238,7 +2215,7 @@ static const char *redis_parse_record(const str *record, parser_arg *root, if (!benc_root) return "failed to decode bencode dictionary"; root->benc = benc_root; - redis_parser = &ng_parser_native; + redis_parser = out->parser = &ng_parser_native; return NULL; } @@ -2375,7 +2352,7 @@ static void json_restore_call(struct redis *r, const str *callid, bool foreign) err = "failed to link maps"; if (json_link_maps(c, &maps, &sfds, root)) goto err8; - if (redis_restore_checkpoints(c, &call, root)) { + if (redis_restore_checkpoints(c, root)) { /* auxiliary state: an unreadable payload disables rollback rather than * discarding an otherwise usable call */ call_checkpoint_free_all(c); @@ -2712,9 +2689,10 @@ static str redis_encode_json(ng_parser_ctx_t *ctx, call_t *c, void **to_free, parser_arg root = parser->dict(ctx); { - parser_arg inner = parser->dict_add_dict(root, "json"); + parser_arg inner = {0}; - { + if (!scope) { + inner = parser->dict_add_dict(root, "json"); JSON_SET_SIMPLE("created","%" PRId64, c->created); JSON_SET_SIMPLE("destroyed","%" PRId64, c->destroyed); JSON_SET_SIMPLE("last_signal","%" PRId64, c->last_signal_us); @@ -2730,13 +2708,6 @@ static str redis_encode_json(ng_parser_ctx_t *ctx, call_t *c, void **to_free, JSON_SET_SIMPLE_STR("recording_metadata", &c->metadata); JSON_SET_SIMPLE("block_dtmf","%i", c->block_dtmf); JSON_SET_SIMPLE("call_flags", "%" PRIu64, atomic64_get_na(&c->call_flags)); - unsigned int num_checkpoints = 0; - if (!scope) { - for (const struct call_checkpoint *cp = c->checkpoints; cp; cp = cp->next) - num_checkpoints++; - if (num_checkpoints) - JSON_SET_SIMPLE("num_checkpoints", "%u", num_checkpoints); - } if (c->created_from.len) JSON_SET_SIMPLE_STR("created_from", &c->created_from); @@ -2752,22 +2723,21 @@ static str redis_encode_json(ng_parser_ctx_t *ctx, call_t *c, void **to_free, JSON_SET_SIMPLE_STR("recording_random_tag", &c->recording_random_tag); } - if (!scope) { - unsigned int ci = 0; - for (const struct call_checkpoint *cp = c->checkpoints; cp; cp = cp->next, ci++) { - snprintf(tmp, sizeof(tmp), "checkpoint-%u", ci); - inner = parser->dict_add_dict_dup(root, tmp); - JSON_SET_SIMPLE("offerer", "%u", cp->offerer->unique_id); - JSON_SET_SIMPLE("answerer", "%u", cp->answerer->unique_id); - JSON_SET_SIMPLE("pending", "%i", cp->pending ? 1 : 0); - if (cp->snapshot.len) { - /* nested as a string; heap buffer rather than a VLA, as escape() can - * need up to 3x the input */ - char *enc = g_malloc_n(cp->snapshot.len + 1, 3); - str encs = parser->escape(enc, cp->snapshot.s, cp->snapshot.len); - parser->dict_add_str_dup(inner, "snapshot", &encs); - g_free(enc); - } + for (__auto_type l = scope ? NULL : c->monologues.head; l; l = l->next) { + const struct call_monologue *ml = l->data; + if (!ml->checkpoint) + continue; + snprintf(tmp, sizeof(tmp), "checkpoint-%u", ml->unique_id); + inner = parser->dict_add_dict_dup(root, tmp); + JSON_SET_SIMPLE("pending", "%i", ml->checkpoint->pending ? 1 : 0); + if (ml->checkpoint->snapshot.len) { + /* nested as a string; heap buffer rather than a VLA, as escape() can + * need up to 3x the input */ + char *enc = g_malloc_n(ml->checkpoint->snapshot.len + 1, 3); + str encs = parser->escape(enc, ml->checkpoint->snapshot.s, + ml->checkpoint->snapshot.len); + parser->dict_add_str_dup(inner, "snapshot", &encs); + g_free(enc); } } @@ -3132,8 +3102,8 @@ static str redis_encode_json(ng_parser_ctx_t *ctx, call_t *c, void **to_free, } -str redis_snapshot_encode(call_t *c, struct call_monologue *a, struct call_monologue *b) { - struct call_monologue *scope[2] = { a, b }; +str redis_snapshot_encode(call_t *c, struct call_monologue *ml) { + struct call_monologue *scope[2] = { ml, ml }; ng_parser_ctx_t ctx; bencode_buffer_t bbuf; // never leaves the daemon, so the format is ours to pick @@ -3439,93 +3409,104 @@ static unsigned int snapshot_medias_len(struct call_monologue *ml, parser_arg ro return n; } -static void snapshot_apply_medias(call_t *c, struct call_monologue **mls, unsigned int n, - parser_arg root) -{ - for (unsigned int i = 0; i < n; i++) { - if (!mls[i]) +static void snapshot_apply_medias(call_t *c, struct call_monologue *ml, parser_arg root) { + for (unsigned int j = 0; j < ml->medias->len; j++) { + struct call_media *m = ml->medias->pdata[j]; + struct redis_hash rh; + if (!m || json_get_hash(&rh, "media", m->unique_id, root)) continue; - for (unsigned int j = 0; j < mls[i]->medias->len; j++) { - struct call_media *m = mls[i]->medias->pdata[j]; - struct redis_hash rh; - if (!m || json_get_hash(&rh, "media", m->unique_id, root)) - continue; - snapshot_apply_media(c, m, &rh, root); - redis_hash_destroy(&rh); - } + snapshot_apply_media(c, m, &rh, root); + redis_hash_destroy(&rh); } } -static void snapshot_apply_monologues(struct call_monologue **mls, unsigned int n, - parser_arg root) -{ - for (unsigned int i = 0; i < n; i++) { - struct call_monologue *ml = mls[i]; - if (!ml) - continue; - struct redis_hash rh; - if (!json_get_hash(&rh, "tag", ml->unique_id, root)) { - snapshot_apply_monologue(ml, &rh); - redis_hash_destroy(&rh); - } - unsigned int keep = snapshot_medias_len(ml, root); - for (unsigned int j = keep; j < ml->medias->len; j++) - call_media_stop(ml->medias->pdata[j]); - if (keep < ml->medias->len) - t_ptr_array_set_size(ml->medias, keep); +static void snapshot_apply_monologues(struct call_monologue *ml, parser_arg root) { + struct redis_hash rh; + if (!json_get_hash(&rh, "tag", ml->unique_id, root)) { + snapshot_apply_monologue(ml, &rh); + redis_hash_destroy(&rh); } + unsigned int keep = snapshot_medias_len(ml, root); + for (unsigned int j = keep; j < ml->medias->len; j++) + call_media_stop(ml->medias->pdata[j]); + if (keep < ml->medias->len) + t_ptr_array_set_size(ml->medias, keep); } -static void snapshot_apply_streams(call_t *c, struct call_monologue **mls, unsigned int n, - parser_arg root) -{ - for (unsigned int i = 0; i < n; i++) { - if (!mls[i]) +static void snapshot_apply_streams(call_t *c, struct call_monologue *ml, parser_arg root) { + for (unsigned int j = 0; j < ml->medias->len; j++) { + struct call_media *m = ml->medias->pdata[j]; + if (!m) continue; - for (unsigned int j = 0; j < mls[i]->medias->len; j++) { - struct call_media *m = mls[i]->medias->pdata[j]; - if (!m) + for (__auto_type l = m->streams.head; l; l = l->next) { + struct packet_stream *ps = l->data; + struct redis_hash rh; + if (json_get_hash(&rh, "stream", ps->unique_id, root)) continue; - for (__auto_type l = m->streams.head; l; l = l->next) { - struct packet_stream *ps = l->data; - struct redis_hash rh; - if (json_get_hash(&rh, "stream", ps->unique_id, root)) - continue; - snapshot_apply_stream(c, ps, &rh, root); - redis_hash_destroy(&rh); - __init_stream(ps); - } + snapshot_apply_stream(c, ps, &rh, root); + redis_hash_destroy(&rh); + __init_stream(ps); } } } -bool redis_snapshot_apply(call_t *c, const str *snap, struct call_monologue *a, - struct call_monologue *b) -{ - if (!snap || !snap->len) - return false; - - struct redis_parsed_record parsed = {0}; - parser_arg root = {0}; +bool redis_snapshot_apply(call_t *c, struct call_monologue *a, struct call_monologue *b) { + struct call_monologue *mls[2] = { a, b }; + struct redis_parsed_record parsed[2] = {0}; + parser_arg root[2] = {0}; + bool live[2] = { false, false }; bool ok = false; - if (redis_parse_record(snap, &root, &parsed)) - goto out; + for (unsigned int i = 0; i < G_N_ELEMENTS(mls); i++) { + struct call_monologue *ml = mls[i]; + if (!ml || !ml->checkpoint || !ml->checkpoint->pending) + continue; + if (!ml->checkpoint->snapshot.len) + continue; + if (redis_parse_record(&ml->checkpoint->snapshot, &root[i], &parsed[i])) + goto out; + live[i] = true; + } + if (!live[0] && !live[1]) + goto out; - struct call_monologue *mls[2] = { a, b }; + // order matters: monologues need the medias, subscriptions need the + // monologues, and initialising the streams needs both + for (unsigned int i = 0; i < G_N_ELEMENTS(mls); i++) { + if (!live[i]) + continue; + redis_parser = parsed[i].parser; + snapshot_apply_medias(c, mls[i], root[i]); + } + for (unsigned int i = 0; i < G_N_ELEMENTS(mls); i++) { + if (!live[i]) + continue; + redis_parser = parsed[i].parser; + snapshot_apply_monologues(mls[i], root[i]); + } - // order matters: monologues need the medias, subscriptions need the monologues, - // and initialising the streams needs both - snapshot_apply_medias(c, mls, G_N_ELEMENTS(mls), root); - snapshot_apply_monologues(mls, G_N_ELEMENTS(mls), root); update_init_monologue_subscribers(a, OP_OFFER); update_init_monologue_subscribers(b, OP_ANSWER); - snapshot_apply_streams(c, mls, G_N_ELEMENTS(mls), root); + + for (unsigned int i = 0; i < G_N_ELEMENTS(mls); i++) { + if (!live[i]) + continue; + redis_parser = parsed[i].parser; + snapshot_apply_streams(c, mls[i], root[i]); + } + + for (unsigned int i = 0; i < G_N_ELEMENTS(mls); i++) { + if (!live[i]) + continue; + redis_snapshot_free(&mls[i]->checkpoint->snapshot); + mls[i]->checkpoint->pending = false; + } ok = true; out: - redis_parsed_record_free(&parsed); + for (unsigned int i = 0; i < G_N_ELEMENTS(mls); i++) + redis_parsed_record_free(&parsed[i]); return ok; } diff --git a/docs/ng_control_protocol.md b/docs/ng_control_protocol.md index 5327dd270..c1a115018 100644 --- a/docs/ng_control_protocol.md +++ b/docs/ng_control_protocol.md @@ -1889,16 +1889,16 @@ contain: NG messages. The successful response contains `rolled-back`, set to `1` if a pending -checkpoint was restored or `0` if there was no matching pending checkpoint. -Repeating a successful rollback is therefore safe and returns `rolled-back: 0`. +checkpoint was restored or `0` if there was none outstanding. Repeating a +successful rollback is therefore safe and returns `rolled-back: 0`. -A dialogue holds at most one outstanding checkpoint. Offers that arrive before -an exchange completes belong to the same uncommitted exchange and keep the -existing snapshot, so a rollback returns to the last completed offer/answer -rather than to an intermediate one. A signalling element should not therefore -issue a new offer for a dialogue while a rollback for it is still in flight: -the rollback restores the last completed state and the newer offer is undone -with it. +Each side of a dialogue holds at most one outstanding checkpoint. Offers that +arrive before an exchange completes belong to the same uncommitted exchange and +keep the existing snapshot, so a rollback returns to the last completed +offer/answer rather than to an intermediate one. A signalling element should +not therefore issue a new offer for a dialogue while a rollback for it is still +in flight: the rollback restores the last completed state and the newer offer is +undone with it. Rollback restores addresses and ports, codecs and payload mappings, transport profile, media direction, and SDES configuration including keys. ICE @@ -1911,6 +1911,11 @@ State the rejected offer introduced is removed as well as overwritten. An offer that upgraded a media to DTLS-SRTP, for example, leaves behind no TLS ID, fingerprint or SRTP context once it has been rolled back. +Where a call has been forked, the offering side is shared between the branches. +Its checkpoint is taken once, before the first uncommitted offer, so rolling +back one branch does not disturb what rolling back another has already +restored. + Sockets and endpoint maps allocated for a rejected offer are not released by a rollback. The media is returned to the sockets it was using, and the surplus is reclaimed with the call. diff --git a/include/call.h b/include/call.h index e692e749a..067437b57 100644 --- a/include/call.h +++ b/include/call.h @@ -669,6 +669,8 @@ struct call_monologue { str moh_file; atomic64 ml_flags; + + struct call_checkpoint *checkpoint; }; TYPED_GHASHTABLE(str_ml_ht, str, struct call_monologue, str_hash, str_equal, NULL, NULL) @@ -821,7 +823,6 @@ struct call { atomic64 call_flags; unsigned int update_iter; unsigned int media_rec_slots; - struct call_checkpoint *checkpoints; }; @@ -962,11 +963,8 @@ void call_media_unkernelize(struct call_media *media, const char *reason); void __monologue_unconfirm(struct call_monologue *monologue, const char *); void __media_unconfirm(struct call_media *media, const char *); __attribute__((nonnull(1))) -/* one dialogue's state from before an offer, held as a call record snapshot */ +/* one monologue's state from before an offer, held as a call record snapshot */ struct call_checkpoint { - struct call_checkpoint *next; - struct call_monologue *offerer; - struct call_monologue *answerer; bool pending; str snapshot; }; diff --git a/include/redis.h b/include/redis.h index 753182f75..e4039e096 100644 --- a/include/redis.h +++ b/include/redis.h @@ -80,9 +80,9 @@ struct redis_list { void **ptrs; }; -str redis_snapshot_encode(call_t *, struct call_monologue *, struct call_monologue *); +str redis_snapshot_encode(call_t *, struct call_monologue *); void redis_snapshot_free(str *); -bool redis_snapshot_apply(call_t *, const str *, struct call_monologue *, struct call_monologue *); +bool redis_snapshot_apply(call_t *, struct call_monologue *, struct call_monologue *); int redis_encode_sdes_params(const ng_parser_t *, parser_arg, const char *, const sdes_q *); void redis_encode_dtls_fingerprint(const ng_parser_t *, parser_arg, diff --git a/t/auto-daemon-tests-rollback-redis.pl b/t/auto-daemon-tests-rollback-redis.pl index c8c669cc8..c6d7a3540 100644 --- a/t/auto-daemon-tests-rollback-redis.pl +++ b/t/auto-daemon-tests-rollback-redis.pl @@ -236,8 +236,9 @@ sub inspect_checkpoint { my $decoded = decode_record($record); my $checkpoint = $decoded->{'checkpoint-0'}; ok(defined $checkpoint, "$redis_format record contains a checkpoint"); - is(field($decoded->{json}{num_checkpoints}), 1, - "$redis_format record contains one dialogue checkpoint"); + # keyed on the monologue id, so both sides of the dialogue carry one + ok(defined $decoded->{'checkpoint-1'}, + "$redis_format both monologues carry a checkpoint"); is(field($checkpoint->{pending}) ? 1 : 0, $expected_pending, "$redis_format pending state serialized"); # The snapshot is a nested call record, so it is checked for presence rather @@ -284,7 +285,7 @@ sub durable_fields { my %call = %{$decoded->{$key}}; # Wall-clock and bookkeeping that moves on its own. delete @call{qw(created created_us created_ts last_signal deleted - ml_deleted last_redis_update num_checkpoints num_sfds num_maps)}; + ml_deleted last_redis_update num_sfds num_maps)}; $out{$key} = \%call; next; } @@ -306,7 +307,8 @@ sub redis_rtpe_req { sub assert_record_without_checkpoint { my ($record) = @_; my $decoded = decode_record($record); - is(field($decoded->{json}{num_checkpoints}) // 0, 0, + my @checkpoints = grep { /^checkpoint-/ } keys %$decoded; + is(scalar(@checkpoints), 0, "$redis_format invalid checkpoint is omitted from the restored call record"); } diff --git a/t/auto-daemon-tests-rollback.pl b/t/auto-daemon-tests-rollback.pl index 5f827ec2d..7fe19868d 100644 --- a/t/auto-daemon-tests-rollback.pl +++ b/t/auto-daemon-tests-rollback.pl @@ -354,10 +354,17 @@ my $fork_a_repeat = rtpe_req('rollback', 'repeat rollback fork A', { 'from-tag' => $fork_from, 'to-tag' => $fork_a, 'via-branch' => 'fork-a', }); is($fork_a_repeat->{'rolled-back'}, 0, 'fork A checkpoint was consumed'); +my $fork_committed = rtpe_req('query', 'fork state before second rollback', {}); my $fork_b_rollback = rtpe_req('rollback', 'rollback fork B', { 'from-tag' => $fork_from, 'to-tag' => $fork_b, 'via-branch' => 'fork-b', }); is($fork_b_rollback->{'rolled-back'}, 1, 'fork B checkpoint remains pending'); +# The caller monologue is shared between branches. Rolling back the second one +# must not reinstate what rolling back the first one undid. +my $fork_after = rtpe_req('query', 'fork state after both rollbacks', {}); +is_deeply($fork_after->{tags}{$fork_from}{medias}, + $fork_committed->{tags}{$fork_from}{medias}, + 'rolling back the second fork leaves the caller alone'); new_call; rtpe_req('offer', 'stress initial offer', {