22884183267

Committed 10 Mar 2026 02:25AM UTC coverage: 58.71%. First build

Build # 22884183267

Build Type

Pull #2097

github

Committed by

web-flow

Commit Message

Merge 8749e3387 into 65bd8cf55

Pull Request Pull Request #2097: feat: support sync_stop_before_deinit in graph

Coverage Stats

165 of 236 new or added lines in 15 files covered. (69.92%)

53330 of 90837 relevant lines covered (58.71%)

806175.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.27

/core/src/ten_runtime/app/msg_interface/start_graph.c

//
// Copyright © 2025 Agora
// This file is part of TEN Framework, an open source project.
// Licensed under the Apache License, Version 2.0, with certain conditions.
// Refer to the "LICENSE" file in the root directory for more information.
//
#include "include_internal/ten_runtime/app/msg_interface/start_graph.h"

#include <time.h>

#include "include_internal/ten_runtime/app/app.h"
#include "include_internal/ten_runtime/app/base_dir.h"
#include "include_internal/ten_runtime/app/close.h"
#include "include_internal/ten_runtime/app/engine_interface.h"
#include "include_internal/ten_runtime/app/metadata.h"
#include "include_internal/ten_runtime/app/msg_interface/common.h"
#include "include_internal/ten_runtime/app/predefined_graph.h"
#include "include_internal/ten_runtime/connection/connection.h"
#include "include_internal/ten_runtime/connection/migration.h"
#include "include_internal/ten_runtime/engine/engine.h"
#include "include_internal/ten_runtime/engine/internal/migration.h"
#include "include_internal/ten_runtime/engine/msg_interface/common.h"
#include "include_internal/ten_runtime/extension/extension_info/extension_info.h"
#include "include_internal/ten_runtime/extension_group/extension_group_info/extension_group_info.h"
#include "include_internal/ten_runtime/msg/cmd_base/cmd/start_graph/cmd.h"
#include "include_internal/ten_runtime/msg/msg.h"
#include "include_internal/ten_runtime/protocol/protocol.h"
#include "include_internal/ten_rust/ten_rust.h"
#include "ten_runtime/app/app.h"
#include "ten_runtime/common/error_code.h"
#include "ten_runtime/msg/msg.h"
#include "ten_utils/lib/error.h"
#include "ten_utils/lib/smart_ptr.h"
#include "ten_utils/lib/string.h"
#include "ten_utils/log/log.h"
#include "ten_utils/macro/check.h"
#include "ten_utils/macro/memory.h"

static bool ten_app_fill_start_graph_cmd_extensions_info_from_predefined_graph(
    ten_app_t *self, ten_shared_ptr_t *cmd, ten_error_t *err) {
  TEN_ASSERT(self, "Should not happen.");
  TEN_ASSERT(ten_app_check_integrity(self, true), "Should not happen.");
  TEN_ASSERT(cmd, "Should not happen.");
  TEN_ASSERT(ten_cmd_base_check_integrity(cmd), "Should not happen.");

  ten_string_t *predefined_graph_name =
      ten_cmd_start_graph_get_predefined_graph_name(cmd);
  if (ten_string_is_empty(predefined_graph_name)) {
    return true;
  }

  ten_list_t *extensions_info = ten_cmd_start_graph_get_extensions_info(cmd);
  ten_list_t *extension_groups_info =
      ten_cmd_start_graph_get_extension_groups_info(cmd);

  bool res = ten_app_get_predefined_graph_extensions_and_groups_info_by_name(
      self, ten_string_get_raw_str(predefined_graph_name), extensions_info,
      extension_groups_info, err);
  TEN_ASSERT(res, "should not happen.");
  if (!res) {
    return false;
  }

  return true;
}

void ten_app_fill_start_graph_cmd_node_app_uri(ten_app_t *self,
                                               ten_shared_ptr_t *cmd) {
  TEN_ASSERT(self, "Should not happen.");
  TEN_ASSERT(ten_app_check_integrity(self, true), "Should not happen.");

  TEN_ASSERT(cmd, "Should not happen.");
  TEN_ASSERT(ten_cmd_base_check_integrity(cmd), "Should not happen.");

  TEN_ASSERT(ten_msg_get_type(cmd) == TEN_MSG_TYPE_CMD_START_GRAPH,
             "Should not happen.");

  ten_list_t *extensions_info = ten_cmd_start_graph_get_extensions_info(cmd);
  ten_list_t *extension_groups_info =
      ten_cmd_start_graph_get_extension_groups_info(cmd);

  ten_extensions_info_fill_app_uri(extensions_info,
                                   ten_string_get_raw_str(&self->uri));
  ten_extension_groups_info_fill_app_uri(extension_groups_info,
                                         ten_string_get_raw_str(&self->uri));
}

bool ten_app_handle_start_graph_cmd(ten_app_t *self,
                                    ten_connection_t *connection,
                                    ten_shared_ptr_t *cmd, ten_error_t *err) {
  TEN_ASSERT(self, "Invalid argument.");
  TEN_ASSERT(ten_app_check_integrity(self, true), "Invalid argument.");
  TEN_ASSERT(cmd, "Invalid argument.");
  TEN_ASSERT(ten_cmd_base_check_integrity(cmd), "Invalid argument.");
  TEN_ASSERT(ten_msg_get_type(cmd) == TEN_MSG_TYPE_CMD_START_GRAPH,
             "Invalid argument.");
  TEN_ASSERT(ten_msg_get_dest_cnt(cmd) == 1, "Invalid argument.");
  TEN_ASSERT(
      connection ? ten_app_has_orphan_connection(self, connection) : true,
      "Invalid argument.");

  // If the start_graph command contains graph_json, we should flatten the
  // graph json first, and then apply the flattened graph json to the cmd.
  ten_string_t *graph_json_str = ten_cmd_start_graph_get_graph_json(cmd);
  if (graph_json_str && !ten_string_is_empty(graph_json_str)) {
#if defined(TEN_ENABLE_TEN_RUST_APIS)
    // Get the cmd source location.
    ten_loc_t *src_loc = ten_msg_get_src_loc(cmd);
    ten_value_t *src_loc_value = ten_loc_to_value(src_loc);

    ten_json_t src_loc_json =
        TEN_JSON_INIT_VAL(ten_json_create_new_ctx(), true);
    bool success = ten_value_to_json(src_loc_value, &src_loc_json);
    if (!success) {
      TEN_LOGE("Failed to convert src loc to json.");
      ten_value_destroy(src_loc_value);
      return false;
    }

    bool must_free = false;
    const char *src_loc_json_str =
        ten_json_to_string(&src_loc_json, NULL, &must_free);
    ten_json_deinit(&src_loc_json);

    if (!src_loc_json_str) {
      TEN_LOGE("Failed to convert src loc to json string.");
      ten_value_destroy(src_loc_value);
      return false;
    }

    ten_value_destroy(src_loc_value);

    // Flatten the graph json string.
    char *err_msg = NULL;
    const char *flattened_graph_json_str =
        ten_rust_graph_validate_complete_flatten(
            ten_string_get_raw_str(graph_json_str), ten_app_get_base_dir(self),
            src_loc_json_str, &err_msg);

    if (must_free) {
      TEN_FREE(src_loc_json_str);
    }

    if (!flattened_graph_json_str) {
      TEN_LOGE("Failed to flatten graph json string: %s", err_msg);
      ten_rust_free_cstring(err_msg);
      return false;
    }

    bool rc = ten_cmd_start_graph_apply_graph_json_str(
        cmd, flattened_graph_json_str, err);
    if (!rc) {
      TEN_LOGE(
          "Failed to apply flattened graph json string to cmd, flattened "
          "graph json string: %s, error: %s",
          flattened_graph_json_str, ten_error_message(err));
      return false;
    }

    ten_rust_free_cstring(flattened_graph_json_str);
#else
    bool rc = ten_cmd_start_graph_apply_graph_json_str(
        cmd, ten_string_get_raw_str(graph_json_str), err);
    if (!rc) {
      TEN_LOGE(
          "Failed to apply graph json string to cmd, graph json string: "
          "%s, error: %s",
          ten_string_get_raw_str(graph_json_str), ten_error_message(err));
      return false;
    }
#endif
  }

  // If the start_graph command is aimed at initting from a predefined graph, we
  // should append the extension info list of the predefined graph to the cmd.
  if (!ten_app_fill_start_graph_cmd_extensions_info_from_predefined_graph(
          self, cmd, err)) {
    return false;
  }

  // Fill the app uri of the nodes in the start_graph cmd.
  ten_app_fill_start_graph_cmd_node_app_uri(self, cmd);

  // sync_stop_before_deinit relies on an in-engine barrier that only covers
  // extensions within one app process.  Reject multi-app graphs upfront so
  // the engine never enters an undefined synchronization state.
  if (ten_cmd_start_graph_get_sync_stop_before_deinit(cmd)) {
    ten_list_t *extensions_info = ten_cmd_start_graph_get_extensions_info(cmd);
    const char *first_app_uri = NULL;

    ten_list_foreach (extensions_info, ext_iter) {
      ten_extension_info_t *ext_info =
          ten_shared_ptr_get_data(ten_smart_ptr_listnode_get(ext_iter.node));
      TEN_ASSERT(ext_info, "Invalid argument.");

      const char *app_uri =
          ten_string_get_raw_str(&ext_info->loc.app_uri);

      if (first_app_uri == NULL) {
        first_app_uri = app_uri;
      } else if (!ten_c_string_is_equal(first_app_uri, app_uri)) {
        ten_error_set(
            err, TEN_ERROR_CODE_INVALID_GRAPH,
            "sync_stop_before_deinit=true requires all nodes in the graph to "
            "belong to the same app, but found nodes from different apps: "
            "'%s' and '%s'",
            first_app_uri, app_uri);
        return false;
      }
    }
  }

  ten_string_t *dest_graph_id = &ten_msg_get_first_dest_loc(cmd)->graph_id;

  ten_engine_t *engine = ten_app_get_engine_by_graph_id(
      self, ten_string_get_raw_str(dest_graph_id));
  if (engine == NULL) {
    // The engine does not exist, create one, and send 'cmd' to the newly
    // created engine.
    engine = ten_app_create_engine(self, cmd);
  } else {
    // The engine of the graph has already been created, this condition would be
    // hit in polygon graph.
  }

  // No matter the situation, it is up to the engine to handle the connect
  // command and return the corresponding cmd result.
  ten_app_do_connection_migration_or_push_to_engine_queue(connection, engine,
                                                          cmd);

  return true;
}

1	//
2	// Copyright © 2025 Agora
3	// This file is part of TEN Framework, an open source project.
4	// Licensed under the Apache License, Version 2.0, with certain conditions.
5	// Refer to the "LICENSE" file in the root directory for more information.
6	//
7	#include "include_internal/ten_runtime/app/msg_interface/start_graph.h"
8
9	#include <time.h>
10
11	#include "include_internal/ten_runtime/app/app.h"
12	#include "include_internal/ten_runtime/app/base_dir.h"
13	#include "include_internal/ten_runtime/app/close.h"
14	#include "include_internal/ten_runtime/app/engine_interface.h"
15	#include "include_internal/ten_runtime/app/metadata.h"
16	#include "include_internal/ten_runtime/app/msg_interface/common.h"
17	#include "include_internal/ten_runtime/app/predefined_graph.h"
18	#include "include_internal/ten_runtime/connection/connection.h"
19	#include "include_internal/ten_runtime/connection/migration.h"
20	#include "include_internal/ten_runtime/engine/engine.h"
21	#include "include_internal/ten_runtime/engine/internal/migration.h"
22	#include "include_internal/ten_runtime/engine/msg_interface/common.h"
23	#include "include_internal/ten_runtime/extension/extension_info/extension_info.h"
24	#include "include_internal/ten_runtime/extension_group/extension_group_info/extension_group_info.h"
25	#include "include_internal/ten_runtime/msg/cmd_base/cmd/start_graph/cmd.h"
26	#include "include_internal/ten_runtime/msg/msg.h"
27	#include "include_internal/ten_runtime/protocol/protocol.h"
28	#include "include_internal/ten_rust/ten_rust.h"
29	#include "ten_runtime/app/app.h"
30	#include "ten_runtime/common/error_code.h"
31	#include "ten_runtime/msg/msg.h"
32	#include "ten_utils/lib/error.h"
33	#include "ten_utils/lib/smart_ptr.h"
34	#include "ten_utils/lib/string.h"
35	#include "ten_utils/log/log.h"
36	#include "ten_utils/macro/check.h"
37	#include "ten_utils/macro/memory.h"
38
39	static bool ten_app_fill_start_graph_cmd_extensions_info_from_predefined_graph(
40	ten_app_t self, ten_shared_ptr_t cmd, ten_error_t *err) {	1,780✔
41	TEN_ASSERT(self, "Should not happen.");	1,780✔
42	TEN_ASSERT(ten_app_check_integrity(self, true), "Should not happen.");	1,780✔
43	TEN_ASSERT(cmd, "Should not happen.");	1,780✔
44	TEN_ASSERT(ten_cmd_base_check_integrity(cmd), "Should not happen.");	1,780✔
45
46	ten_string_t *predefined_graph_name =	1,780✔
47	ten_cmd_start_graph_get_predefined_graph_name(cmd);	1,780✔
48	if (ten_string_is_empty(predefined_graph_name)) {	1,780✔
49	return true;	1,772✔
50	}	1,772✔
51
52	ten_list_t *extensions_info = ten_cmd_start_graph_get_extensions_info(cmd);	8✔
53	ten_list_t *extension_groups_info =	8✔
54	ten_cmd_start_graph_get_extension_groups_info(cmd);	8✔
55
56	bool res = ten_app_get_predefined_graph_extensions_and_groups_info_by_name(	8✔
57	self, ten_string_get_raw_str(predefined_graph_name), extensions_info,	8✔
58	extension_groups_info, err);	8✔
59	TEN_ASSERT(res, "should not happen.");	8✔
60	if (!res) {	8✔
61	return false;	×
62	}	×
63
64	return true;	8✔
65	}	8✔
66
67	void ten_app_fill_start_graph_cmd_node_app_uri(ten_app_t *self,
68	ten_shared_ptr_t *cmd) {	1,780✔
69	TEN_ASSERT(self, "Should not happen.");	1,780✔
70	TEN_ASSERT(ten_app_check_integrity(self, true), "Should not happen.");	1,780✔
71
72	TEN_ASSERT(cmd, "Should not happen.");	1,780✔
73	TEN_ASSERT(ten_cmd_base_check_integrity(cmd), "Should not happen.");	1,780✔
74
75	TEN_ASSERT(ten_msg_get_type(cmd) == TEN_MSG_TYPE_CMD_START_GRAPH,	1,780✔
76	"Should not happen.");	1,780✔
77
78	ten_list_t *extensions_info = ten_cmd_start_graph_get_extensions_info(cmd);	1,780✔
79	ten_list_t *extension_groups_info =	1,780✔
80	ten_cmd_start_graph_get_extension_groups_info(cmd);	1,780✔
81
82	ten_extensions_info_fill_app_uri(extensions_info,	1,780✔
83	ten_string_get_raw_str(&self->uri));	1,780✔
84	ten_extension_groups_info_fill_app_uri(extension_groups_info,	1,780✔
85	ten_string_get_raw_str(&self->uri));	1,780✔
86	}	1,780✔
87
88	bool ten_app_handle_start_graph_cmd(ten_app_t *self,
89	ten_connection_t *connection,
90	ten_shared_ptr_t cmd, ten_error_t err) {	1,780✔
91	TEN_ASSERT(self, "Invalid argument.");	1,780✔
92	TEN_ASSERT(ten_app_check_integrity(self, true), "Invalid argument.");	1,780✔
93	TEN_ASSERT(cmd, "Invalid argument.");	1,780✔
94	TEN_ASSERT(ten_cmd_base_check_integrity(cmd), "Invalid argument.");	1,780✔
95	TEN_ASSERT(ten_msg_get_type(cmd) == TEN_MSG_TYPE_CMD_START_GRAPH,	1,780✔
96	"Invalid argument.");	1,780✔
97	TEN_ASSERT(ten_msg_get_dest_cnt(cmd) == 1, "Invalid argument.");	1,780✔
98	TEN_ASSERT(	1,780✔
99	connection ? ten_app_has_orphan_connection(self, connection) : true,	1,780✔
100	"Invalid argument.");	1,780✔
101
102	// If the start_graph command contains graph_json, we should flatten the
103	// graph json first, and then apply the flattened graph json to the cmd.
104	ten_string_t *graph_json_str = ten_cmd_start_graph_get_graph_json(cmd);	1,780✔
105	if (graph_json_str && !ten_string_is_empty(graph_json_str)) {	1,780✔
106	#if defined(TEN_ENABLE_TEN_RUST_APIS)	1,027✔
107	// Get the cmd source location.
108	ten_loc_t *src_loc = ten_msg_get_src_loc(cmd);	1,027✔
109	ten_value_t *src_loc_value = ten_loc_to_value(src_loc);	1,027✔
110
111	ten_json_t src_loc_json =	1,027✔
112	TEN_JSON_INIT_VAL(ten_json_create_new_ctx(), true);	1,027✔
113	bool success = ten_value_to_json(src_loc_value, &src_loc_json);	1,027✔
114	if (!success) {	1,027✔
115	TEN_LOGE("Failed to convert src loc to json.");	×
116	ten_value_destroy(src_loc_value);	×
117	return false;	×
118	}	×
119
120	bool must_free = false;	1,027✔
121	const char *src_loc_json_str =	1,027✔
122	ten_json_to_string(&src_loc_json, NULL, &must_free);	1,027✔
123	ten_json_deinit(&src_loc_json);	1,027✔
124
125	if (!src_loc_json_str) {	1,027✔
126	TEN_LOGE("Failed to convert src loc to json string.");	×
127	ten_value_destroy(src_loc_value);	×
128	return false;	×
129	}	×
130
131	ten_value_destroy(src_loc_value);	1,027✔
132
133	// Flatten the graph json string.
134	char *err_msg = NULL;	1,027✔
135	const char *flattened_graph_json_str =	1,027✔
136	ten_rust_graph_validate_complete_flatten(	1,027✔
137	ten_string_get_raw_str(graph_json_str), ten_app_get_base_dir(self),	1,027✔
138	src_loc_json_str, &err_msg);	1,027✔
139
140	if (must_free) {	1,027✔
141	TEN_FREE(src_loc_json_str);	1,027✔
142	}	1,027✔
143
144	if (!flattened_graph_json_str) {	1,027✔
145	TEN_LOGE("Failed to flatten graph json string: %s", err_msg);	×
146	ten_rust_free_cstring(err_msg);	×
147	return false;	×
148	}	×
149
150	bool rc = ten_cmd_start_graph_apply_graph_json_str(	1,027✔
151	cmd, flattened_graph_json_str, err);	1,027✔
152	if (!rc) {	1,027✔
153	TEN_LOGE(	×
154	"Failed to apply flattened graph json string to cmd, flattened "	×
155	"graph json string: %s, error: %s",	×
156	flattened_graph_json_str, ten_error_message(err));	×
157	return false;	×
158	}	×
159
160	ten_rust_free_cstring(flattened_graph_json_str);	1,027✔
161	#else
162	bool rc = ten_cmd_start_graph_apply_graph_json_str(
163	cmd, ten_string_get_raw_str(graph_json_str), err);
164	if (!rc) {
165	TEN_LOGE(
166	"Failed to apply graph json string to cmd, graph json string: "
167	"%s, error: %s",
168	ten_string_get_raw_str(graph_json_str), ten_error_message(err));
169	return false;
170	}
171	#endif
172	}	1,027✔
173
174	// If the start_graph command is aimed at initting from a predefined graph, we
175	// should append the extension info list of the predefined graph to the cmd.
176	if (!ten_app_fill_start_graph_cmd_extensions_info_from_predefined_graph(	1,780✔
177	self, cmd, err)) {	1,780✔
178	return false;	×
179	}	×
180
181	// Fill the app uri of the nodes in the start_graph cmd.
182	ten_app_fill_start_graph_cmd_node_app_uri(self, cmd);	1,780✔
183
184	// sync_stop_before_deinit relies on an in-engine barrier that only covers
185	// extensions within one app process. Reject multi-app graphs upfront so
186	// the engine never enters an undefined synchronization state.
187	if (ten_cmd_start_graph_get_sync_stop_before_deinit(cmd)) {	1,780✔
188	ten_list_t *extensions_info = ten_cmd_start_graph_get_extensions_info(cmd);	2✔
189	const char *first_app_uri = NULL;	2✔
190
191	ten_list_foreach (extensions_info, ext_iter) {	5✔
192	ten_extension_info_t *ext_info =	5✔
193	ten_shared_ptr_get_data(ten_smart_ptr_listnode_get(ext_iter.node));	5✔
194	TEN_ASSERT(ext_info, "Invalid argument.");	5✔
195
196	const char *app_uri =	5✔
197	ten_string_get_raw_str(&ext_info->loc.app_uri);	5✔
198
199	if (first_app_uri == NULL) {	5✔
200	first_app_uri = app_uri;	2✔
201	} else if (!ten_c_string_is_equal(first_app_uri, app_uri)) {	3✔
NEW 202	ten_error_set(	×
NEW 203	err, TEN_ERROR_CODE_INVALID_GRAPH,	×
NEW 204	"sync_stop_before_deinit=true requires all nodes in the graph to "	×
NEW 205	"belong to the same app, but found nodes from different apps: "	×
NEW 206	"'%s' and '%s'",	×
NEW 207	first_app_uri, app_uri);	×
NEW 208	return false;	×
NEW 209	}	×
210	}	5✔
211	}	2✔
212
213	ten_string_t *dest_graph_id = &ten_msg_get_first_dest_loc(cmd)->graph_id;	1,780✔
214
215	ten_engine_t *engine = ten_app_get_engine_by_graph_id(	1,780✔
216	self, ten_string_get_raw_str(dest_graph_id));	1,780✔
217	if (engine == NULL) {	1,780✔
218	// The engine does not exist, create one, and send 'cmd' to the newly
219	// created engine.
220	engine = ten_app_create_engine(self, cmd);	1,778✔
221	} else {	1,778✔
222	// The engine of the graph has already been created, this condition would be
223	// hit in polygon graph.
224	}	2✔
225
226	// No matter the situation, it is up to the engine to handle the connect
227	// command and return the corresponding cmd result.
228	ten_app_do_connection_migration_or_push_to_engine_queue(connection, engine,	1,780✔
229	cmd);	1,780✔
230
231	return true;
232	}	1,780✔

TEN-framework / ten-framework / 22884183267

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous