This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 0facf7b7e9 fix(admin): enforce unique instance identity across
concurrent writers (#7885)
0facf7b7e9 is described below
commit 0facf7b7e9cc9babb17e4a30400b6029ae86444c
Author: cloudbase-ai <[email protected]>
AuthorDate: Wed Oct 21 10:18:50 2026 +0800
fix(admin): enforce unique instance identity across concurrent writers
(#7885)
Signed-off-by: cloudbase-ai <[email protected]>
Co-authored-by: cloudbase-ai
<[email protected]>
---
.github/scripts/native-sql-matrix.py | 66 ++++++
.github/scripts/test-native-sql-matrix.py | 32 +++
.github/workflows/native-sql-matrix.yml | 2 +-
db/init/mysql/schema.sql | 3 +-
db/init/oracle/schema.sql | 3 +-
db/init/pg/create-table.sql | 3 +-
db/upgrade/2.7.2-upgrade-2.7.3-h2.sql | 42 ++++
db/upgrade/2.7.2-upgrade-2.7.3-mysql.sql | 21 ++
db/upgrade/2.7.2-upgrade-2.7.3-ob.sql | 21 ++
db/upgrade/2.7.2-upgrade-2.7.3-og.sql | 23 ++
db/upgrade/2.7.2-upgrade-2.7.3-oracle.sql | 23 ++
db/upgrade/2.7.2-upgrade-2.7.3-pg.sql | 23 ++
db/upgrade/native-sql-matrix.md | 9 +
db/upgrade/upgrade-guide.md | 31 +++
.../service/impl/InstanceInfoServiceImpl.java | 15 +-
.../resources/mappers/instance-info-sqlmap.xml | 1 +
.../src/main/resources/sql-script/h2/schema.sql | 3 +-
.../admin/mapper/InstanceInfoMapperTest.java | 7 +-
.../admin/mapper/InstanceInfoUpsertTest.java | 247 +++++++++++++++++++++
.../service/impl/InstanceInfoServiceImplTest.java | 22 ++
20 files changed, 589 insertions(+), 8 deletions(-)
diff --git a/.github/scripts/native-sql-matrix.py
b/.github/scripts/native-sql-matrix.py
index 2567729d28..570cc7b00a 100644
--- a/.github/scripts/native-sql-matrix.py
+++ b/.github/scripts/native-sql-matrix.py
@@ -449,6 +449,68 @@ class Engine:
raise AssertionError("Duplicate data permission was accepted")
+ def insert_instance_observations(self, duplicates=False):
+ # Exercise every winner tie-break and every identity component
independently.
+ rows = [
+ ("heartbeat", "ns", "ip", "port", "type", 200, 2, 1),
+ ("updated", "ns-updated", "ip", "port", "type", 0, 2, 1),
+ ("created", "ns-created", "ip", "port", "type", 0, 2, 2),
+ ("tie-z", "ns-tie", "ip", "port", "type", 0, 2, 1),
+ ("namespace", "ns-other", "ip", "port", "type", 50, 2, 1),
+ ("ip", "ns", "ip-other", "port", "type", 50, 2, 1),
+ ("port", "ns", "ip", "port-other", "type", 50, 2, 1),
+ ("type", "ns", "ip", "port", "type-other", 50, 2, 1),
+ ("max", "n" * 50, "i" * 128, "p" * 128, "t" * 128, 50, 2, 1),
+ ]
+ if duplicates:
+ rows += [
+ ("old-heartbeat", "ns", "ip", "port", "type", 100, 3, 1),
+ ("old-updated", "ns-updated", "ip", "port", "type", 0, 1, 1),
+ ("old-created", "ns-created", "ip", "port", "type", 0, 2, 1),
+ ("tie-a", "ns-tie", "ip", "port", "type", 0, 2, 1),
+ ]
+ self.sql("\n".join(
+ "INSERT INTO instance_info
(id,namespace_id,instance_ip,instance_port,instance_type,"
+
"instance_info,instance_state,last_heart_beat_time,date_updated,date_created) "
+ f"VALUES
('sql-instance-{identifier}','{namespace}','{ip}','{port}','{kind}','{identifier}',1,{heartbeat},"
+ f"TIMESTAMP '2026-01-0{updated} 00:00:00',TIMESTAMP
'2026-01-0{created} 00:00:00');"
+ for identifier, namespace, ip, port, kind, heartbeat, updated,
created in rows))
+
+ def check_instance_identity(self, indexes, mapper):
+ index = "uk_instance_info_identity"
+ if indexes.get(("instance_info", index)) != ["namespace_id",
"instance_ip", "instance_port", "instance_type"]:
+ raise AssertionError("Wrong instance identity index columns")
+ identifiers = set(self.sql("SELECT id FROM instance_info WHERE id LIKE
'sql-instance-%';").split())
+ expected = {"heartbeat", "updated", "created", "tie-z", "namespace",
"ip", "port", "type", "max"}
+ if identifiers != {"sql-instance-" + name for name in expected}:
+ raise AssertionError("Instance migration lost distinct identities
or retained the wrong observation")
+ try:
+ self.sql("INSERT INTO instance_info
(id,namespace_id,instance_ip,instance_port,instance_type,instance_info,instance_state)
"
+ "VALUES
('sql-instance-conflict','ns','ip','port','type','conflict',2);")
+ except RuntimeError as error:
+ if index not in str(error).lower():
+ raise
+ else:
+ raise AssertionError("Duplicate instance identity was accepted")
+ document = ET.parse(mapper).getroot()
+ statement =
"".join(document.find("update[@id='updateById']").itertext())
+ values = {"id": "'sql-instance-heartbeat'", "dateUpdated": "TIMESTAMP
'2026-02-01 00:00:00'",
+ "instanceIp": "'ip'", "instancePort": "'port'",
"instanceType": "'type'",
+ "instanceInfo": "'updated-info'", "instanceState": "2"}
+ for heartbeat, expected_heartbeat in ((100, 200), (0, 200), (300,
300), (300, 300)):
+ values["lastHeartBeatTime"] = str(heartbeat)
+ sql = re.sub(r"#\{(\w+)[^}]*}", lambda match: values[match[1]],
statement)
+ self.sql(sql + ";")
+ if scalar(self.sql("SELECT last_heart_beat_time FROM instance_info
WHERE id='sql-instance-heartbeat';")) != str(expected_heartbeat):
+ raise AssertionError("Instance update overwrote a newer
heartbeat or lost a newer report")
+ state = "1" if heartbeat < 200 else "2"
+ if scalar(self.sql("SELECT instance_state FROM instance_info WHERE
id='sql-instance-heartbeat';")) != state:
+ raise AssertionError("Instance state does not match the
retained heartbeat")
+ if scalar(self.sql("SELECT COUNT(*) FROM instance_info WHERE
id='sql-instance-heartbeat' "
+ "AND date_created=TIMESTAMP '2026-01-01
00:00:00';")) != "1":
+ raise AssertionError("Instance update changed the canonical row
creation time")
+
+
def execute_matrix(dialect, root, output):
output.mkdir(parents=True, exist_ok=True)
listing, detail = fragment_columns(root /
"shenyu-admin/src/main/resources/mappers/plugin-sqlmap.xml")
@@ -470,10 +532,13 @@ def execute_matrix(dialect, root, output):
if flow == "upgrade":
engine.insert_data_permissions(duplicates=True)
for step in MIGRATIONS_FROM_BASELINE:
+ if step == "2.7.2-upgrade-2.7.3":
+ engine.insert_instance_observations(duplicates=True)
migration = root / f"db/upgrade/{step}-{dialect}.sql"
engine.sql(migration.read_text(encoding="utf-8"))
else:
engine.insert_data_permissions()
+ engine.insert_instance_observations()
engine.check_data(listing, detail)
current_listing = engine.sql("SELECT " + ",".join(listing) + f"
FROM plugin WHERE id='{SENTINEL}';")
if current_listing != original_listing:
@@ -484,6 +549,7 @@ def execute_matrix(dialect, root, output):
raise AssertionError(f"Migration removed rows from
{table}: {sorted(identifiers - after)}")
indexes = engine.indexes()
engine.check_data_permission_uniqueness(indexes)
+ engine.check_instance_identity(indexes, root /
"shenyu-admin/src/main/resources/mappers/instance-info-sqlmap.xml")
if flow == "fresh" and dialect in ("mysql", "ob"):
engine.check_fresh_mysql_schema(schema, indexes)
if flow == "fresh" and dialect in ("pg", "og"):
diff --git a/.github/scripts/test-native-sql-matrix.py
b/.github/scripts/test-native-sql-matrix.py
index a07291f2f9..e9b1263569 100644
--- a/.github/scripts/test-native-sql-matrix.py
+++ b/.github/scripts/test-native-sql-matrix.py
@@ -257,6 +257,38 @@ class NativeSqlMatrixTest(unittest.TestCase):
self.assertRaisesRegex(AssertionError, "lost distinct
grants or retained duplicate"):
engine.check_data_permission_uniqueness(indexes)
+ def
test_instance_identity_check_rejects_missing_uniqueness_and_stale_updates(self):
+ indexes = {("instance_info", "uk_instance_info_identity"):
["namespace_id", "instance_ip", "instance_port", "instance_type"]}
+ names = "heartbeat updated created tie-z namespace ip port type
max".split()
+ mapper = ROOT /
"shenyu-admin/src/main/resources/mappers/instance-info-sqlmap.xml"
+ for dialect in matrix.DIALECTS:
+ for mode in ("success", "missing-key", "wrong-winner",
"stale-overwrite", "unrelated-error"):
+ heartbeats = iter((100, 200, 300, 300) if mode ==
"stale-overwrite" else (200, 200, 300, 300))
+ states = iter((1, 1, 2, 2))
+
+ def sql(query):
+ if query.startswith("SELECT id"):
+ kept = names + (["old-heartbeat"] if mode ==
"wrong-winner" else [])
+ return "\n".join("sql-instance-" + name for name in
kept)
+ if query.startswith("INSERT"):
+ if mode == "missing-key":
+ return ""
+ raise RuntimeError("other error" if mode ==
"unrelated-error" else "duplicate UK_INSTANCE_INFO_IDENTITY")
+ if query.startswith("SELECT last_heart"):
+ return str(next(heartbeats))
+ if query.startswith("SELECT instance_state"):
+ return str(next(states))
+ return "1"
+
+ with self.subTest(dialect=dialect, mode=mode),
tempfile.TemporaryDirectory() as directory:
+ engine = matrix.Engine(dialect, "upgrade", Path(directory))
+ with patch.object(engine, "sql", side_effect=sql):
+ if mode == "success":
+ engine.check_instance_identity(indexes, mapper)
+ else:
+ with self.assertRaises(RuntimeError if mode ==
"unrelated-error" else AssertionError):
+ engine.check_instance_identity(indexes, mapper)
+
def test_actual_mapper_projects_metadata_and_preserves_detail_jar(self):
listing, detail = matrix.fragment_columns(ROOT /
"shenyu-admin/src/main/resources/mappers/plugin-sqlmap.xml")
self.assertNotIn("plugin_jar", listing)
diff --git a/.github/workflows/native-sql-matrix.yml
b/.github/workflows/native-sql-matrix.yml
index 5c5477d2a6..7c97f8797e 100644
--- a/.github/workflows/native-sql-matrix.yml
+++ b/.github/workflows/native-sql-matrix.yml
@@ -77,7 +77,7 @@ jobs:
- name: Verify list/detail/export and index contracts
run: >-
exec bash .github/scripts/maven-with-retry.sh ./mvnw -B -ntp -pl
shenyu-admin -am test
-
-Dtest=PluginMapperTest,PluginServiceTest,PluginControllerTest,AdminQueryIndexTest,PluginIndexUpgradeContractTest,DataPermissionMapperTest,PermissionListenerIntegrationTest
+
-Dtest=PluginMapperTest,PluginServiceTest,PluginControllerTest,AdminQueryIndexTest,PluginIndexUpgradeContractTest,DataPermissionMapperTest,PermissionListenerIntegrationTest,InstanceInfoUpsertTest
-Dsurefire.failIfNoSpecifiedTests=false -Dmaven.javadoc.skip=true
native-sql:
diff --git a/db/init/mysql/schema.sql b/db/init/mysql/schema.sql
index 6465f0246f..b3c78a36ad 100644
--- a/db/init/mysql/schema.sql
+++ b/db/init/mysql/schema.sql
@@ -2379,7 +2379,8 @@ CREATE TABLE IF NOT EXISTS `instance_info` (
`date_created` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT
'date_created',
`date_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP COMMENT 'date_updated',
`last_heart_beat_time` bigint NOT NULL DEFAULT 0 COMMENT 'last heart
beat time',
- PRIMARY KEY (`id`)
+ PRIMARY KEY (`id`),
+ CONSTRAINT uk_instance_info_identity UNIQUE (namespace_id, instance_ip,
instance_port, instance_type)
);
DROP TABLE IF EXISTS `registry_config`;
diff --git a/db/init/oracle/schema.sql b/db/init/oracle/schema.sql
index ec385b4c89..543a0a0cc9 100644
--- a/db/init/oracle/schema.sql
+++ b/db/init/oracle/schema.sql
@@ -2952,7 +2952,8 @@ CREATE TABLE instance_info (
date_created timestamp(3) DEFAULT SYSTIMESTAMP NOT NULL,
date_updated timestamp(3) DEFAULT SYSTIMESTAMP NOT NULL,
last_heart_beat_time number DEFAULT 0 NOT NULL,
- CONSTRAINT instance_info_pk PRIMARY KEY (id)
+ CONSTRAINT instance_info_pk PRIMARY KEY (id),
+ CONSTRAINT uk_instance_info_identity UNIQUE (namespace_id, instance_ip,
instance_port, instance_type)
);
COMMENT ON TABLE instance_info IS 'Instance information table';
diff --git a/db/init/pg/create-table.sql b/db/init/pg/create-table.sql
index d0aadfc4ff..7f6179e235 100644
--- a/db/init/pg/create-table.sql
+++ b/db/init/pg/create-table.sql
@@ -2693,7 +2693,8 @@ CREATE TABLE "public"."instance_info" (
"date_created" timestamp(3) NOT NULL DEFAULT timezone('UTC-8'::text,
(now())::timestamp(0) without time zone),
"date_updated" timestamp(3) NOT NULL DEFAULT timezone('UTC-8'::text,
(now())::timestamp(0) without time zone),
"last_heart_beat_time" int8 NOT NULL DEFAULT 0,
- PRIMARY KEY ("id")
+ PRIMARY KEY ("id"),
+ CONSTRAINT uk_instance_info_identity UNIQUE (namespace_id, instance_ip,
instance_port, instance_type)
);
COMMENT ON COLUMN "public"."instance_info"."id" IS 'primary key';
COMMENT ON COLUMN "public"."instance_info"."namespace_id" IS 'namespace_id';
diff --git a/db/upgrade/2.7.2-upgrade-2.7.3-h2.sql
b/db/upgrade/2.7.2-upgrade-2.7.3-h2.sql
new file mode 100644
index 0000000000..541a12baf7
--- /dev/null
+++ b/db/upgrade/2.7.2-upgrade-2.7.3-h2.sql
@@ -0,0 +1,42 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+-- this file works for H2.
+ALTER TABLE selector RENAME COLUMN loged TO logged;
+ALTER TABLE rule RENAME COLUMN loged TO logged;
+
+-- Stop all Admin writers and back up instance_info before this one-time
migration.
+-- Retain the latest heartbeat, then latest update/create timestamp, then
greatest id.
+-- Rows upgraded from 2.7.1 have heartbeat 0 and are ordered by timestamps and
id.
+DELETE FROM instance_info duplicate_instance
+WHERE EXISTS (
+ SELECT 1 FROM instance_info retained_instance WHERE
+ retained_instance.namespace_id = duplicate_instance.namespace_id
+ AND retained_instance.instance_ip = duplicate_instance.instance_ip
+ AND retained_instance.instance_port = duplicate_instance.instance_port
+ AND retained_instance.instance_type = duplicate_instance.instance_type
+ AND (retained_instance.last_heart_beat_time >
duplicate_instance.last_heart_beat_time
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated >
duplicate_instance.date_updated)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created >
duplicate_instance.date_created)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created =
duplicate_instance.date_created
+ AND retained_instance.id > duplicate_instance.id))
+);
+CREATE UNIQUE INDEX uk_instance_info_identity ON instance_info (namespace_id,
instance_ip, instance_port, instance_type);
diff --git a/db/upgrade/2.7.2-upgrade-2.7.3-mysql.sql
b/db/upgrade/2.7.2-upgrade-2.7.3-mysql.sql
index b0a7fdc256..097c3c8f83 100644
--- a/db/upgrade/2.7.2-upgrade-2.7.3-mysql.sql
+++ b/db/upgrade/2.7.2-upgrade-2.7.3-mysql.sql
@@ -17,3 +17,24 @@
-- this file works for MySQL.
ALTER TABLE `selector` RENAME COLUMN `loged` TO `logged`;
ALTER TABLE `rule` RENAME COLUMN `loged` TO `logged`;
+
+-- Stop all Admin writers and back up instance_info before this one-time
migration.
+-- Retain the latest heartbeat, then latest update/create timestamp, then
greatest id.
+-- Rows upgraded from 2.7.1 have heartbeat 0 and are ordered by timestamps and
id.
+DELETE duplicate_instance FROM instance_info duplicate_instance
+JOIN instance_info retained_instance ON
+ retained_instance.namespace_id = duplicate_instance.namespace_id
+ AND retained_instance.instance_ip = duplicate_instance.instance_ip
+ AND retained_instance.instance_port = duplicate_instance.instance_port
+ AND retained_instance.instance_type = duplicate_instance.instance_type
+ AND (retained_instance.last_heart_beat_time >
duplicate_instance.last_heart_beat_time
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated >
duplicate_instance.date_updated)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created >
duplicate_instance.date_created)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created =
duplicate_instance.date_created
+ AND retained_instance.id > duplicate_instance.id));
+CREATE UNIQUE INDEX uk_instance_info_identity ON instance_info (namespace_id,
instance_ip, instance_port, instance_type);
diff --git a/db/upgrade/2.7.2-upgrade-2.7.3-ob.sql
b/db/upgrade/2.7.2-upgrade-2.7.3-ob.sql
index cc0d06bc94..06090fd91d 100644
--- a/db/upgrade/2.7.2-upgrade-2.7.3-ob.sql
+++ b/db/upgrade/2.7.2-upgrade-2.7.3-ob.sql
@@ -17,3 +17,24 @@
-- this file works for OceanBase (MySQL mode).
ALTER TABLE `selector` RENAME COLUMN `loged` TO `logged`;
ALTER TABLE `rule` RENAME COLUMN `loged` TO `logged`;
+
+-- Stop all Admin writers and back up instance_info before this one-time
migration.
+-- Retain the latest heartbeat, then latest update/create timestamp, then
greatest id.
+-- Rows upgraded from 2.7.1 have heartbeat 0 and are ordered by timestamps and
id.
+DELETE duplicate_instance FROM instance_info duplicate_instance
+JOIN instance_info retained_instance ON
+ retained_instance.namespace_id = duplicate_instance.namespace_id
+ AND retained_instance.instance_ip = duplicate_instance.instance_ip
+ AND retained_instance.instance_port = duplicate_instance.instance_port
+ AND retained_instance.instance_type = duplicate_instance.instance_type
+ AND (retained_instance.last_heart_beat_time >
duplicate_instance.last_heart_beat_time
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated >
duplicate_instance.date_updated)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created >
duplicate_instance.date_created)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created =
duplicate_instance.date_created
+ AND retained_instance.id > duplicate_instance.id));
+CREATE UNIQUE INDEX uk_instance_info_identity ON instance_info (namespace_id,
instance_ip, instance_port, instance_type);
diff --git a/db/upgrade/2.7.2-upgrade-2.7.3-og.sql
b/db/upgrade/2.7.2-upgrade-2.7.3-og.sql
index 2709542ccc..58657c681e 100644
--- a/db/upgrade/2.7.2-upgrade-2.7.3-og.sql
+++ b/db/upgrade/2.7.2-upgrade-2.7.3-og.sql
@@ -17,3 +17,26 @@
-- this file works for openGauss.
ALTER TABLE "public"."selector" RENAME COLUMN "loged" TO "logged";
ALTER TABLE "public"."rule" RENAME COLUMN "loged" TO "logged";
+
+-- Stop all Admin writers and back up instance_info before this one-time
migration.
+-- Retain the latest heartbeat, then latest update/create timestamp, then
greatest id.
+-- Rows upgraded from 2.7.1 have heartbeat 0 and are ordered by timestamps and
id.
+DELETE FROM instance_info duplicate_instance
+WHERE EXISTS (
+ SELECT 1 FROM instance_info retained_instance WHERE
+ retained_instance.namespace_id = duplicate_instance.namespace_id
+ AND retained_instance.instance_ip = duplicate_instance.instance_ip
+ AND retained_instance.instance_port = duplicate_instance.instance_port
+ AND retained_instance.instance_type = duplicate_instance.instance_type
+ AND (retained_instance.last_heart_beat_time >
duplicate_instance.last_heart_beat_time
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated >
duplicate_instance.date_updated)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created >
duplicate_instance.date_created)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created =
duplicate_instance.date_created
+ AND retained_instance.id > duplicate_instance.id))
+);
+CREATE UNIQUE INDEX uk_instance_info_identity ON instance_info (namespace_id,
instance_ip, instance_port, instance_type);
diff --git a/db/upgrade/2.7.2-upgrade-2.7.3-oracle.sql
b/db/upgrade/2.7.2-upgrade-2.7.3-oracle.sql
index a6cfbafda0..b730778015 100644
--- a/db/upgrade/2.7.2-upgrade-2.7.3-oracle.sql
+++ b/db/upgrade/2.7.2-upgrade-2.7.3-oracle.sql
@@ -17,3 +17,26 @@
-- this file works for Oracle.
ALTER TABLE selector RENAME COLUMN loged TO logged;
ALTER TABLE rule RENAME COLUMN loged TO logged;
+
+-- Stop all Admin writers and back up instance_info before this one-time
migration.
+-- Retain the latest heartbeat, then latest update/create timestamp, then
greatest id.
+-- Rows upgraded from 2.7.1 have heartbeat 0 and are ordered by timestamps and
id.
+DELETE FROM instance_info duplicate_instance
+WHERE EXISTS (
+ SELECT 1 FROM instance_info retained_instance WHERE
+ retained_instance.namespace_id = duplicate_instance.namespace_id
+ AND retained_instance.instance_ip = duplicate_instance.instance_ip
+ AND retained_instance.instance_port = duplicate_instance.instance_port
+ AND retained_instance.instance_type = duplicate_instance.instance_type
+ AND (retained_instance.last_heart_beat_time >
duplicate_instance.last_heart_beat_time
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated >
duplicate_instance.date_updated)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created >
duplicate_instance.date_created)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created =
duplicate_instance.date_created
+ AND retained_instance.id > duplicate_instance.id))
+);
+CREATE UNIQUE INDEX uk_instance_info_identity ON instance_info (namespace_id,
instance_ip, instance_port, instance_type);
diff --git a/db/upgrade/2.7.2-upgrade-2.7.3-pg.sql
b/db/upgrade/2.7.2-upgrade-2.7.3-pg.sql
index a63fb80b85..375740e936 100644
--- a/db/upgrade/2.7.2-upgrade-2.7.3-pg.sql
+++ b/db/upgrade/2.7.2-upgrade-2.7.3-pg.sql
@@ -17,3 +17,26 @@
-- this file works for PostgreSQL.
ALTER TABLE "public"."selector" RENAME COLUMN "loged" TO "logged";
ALTER TABLE "public"."rule" RENAME COLUMN "loged" TO "logged";
+
+-- Stop all Admin writers and back up instance_info before this one-time
migration.
+-- Retain the latest heartbeat, then latest update/create timestamp, then
greatest id.
+-- Rows upgraded from 2.7.1 have heartbeat 0 and are ordered by timestamps and
id.
+DELETE FROM instance_info duplicate_instance
+WHERE EXISTS (
+ SELECT 1 FROM instance_info retained_instance WHERE
+ retained_instance.namespace_id = duplicate_instance.namespace_id
+ AND retained_instance.instance_ip = duplicate_instance.instance_ip
+ AND retained_instance.instance_port = duplicate_instance.instance_port
+ AND retained_instance.instance_type = duplicate_instance.instance_type
+ AND (retained_instance.last_heart_beat_time >
duplicate_instance.last_heart_beat_time
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated >
duplicate_instance.date_updated)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created >
duplicate_instance.date_created)
+ OR (retained_instance.last_heart_beat_time =
duplicate_instance.last_heart_beat_time
+ AND retained_instance.date_updated =
duplicate_instance.date_updated
+ AND retained_instance.date_created =
duplicate_instance.date_created
+ AND retained_instance.id > duplicate_instance.id))
+);
+CREATE UNIQUE INDEX uk_instance_info_identity ON instance_info (namespace_id,
instance_ip, instance_port, instance_type);
diff --git a/db/upgrade/native-sql-matrix.md b/db/upgrade/native-sql-matrix.md
index a702559ec2..c69c789aa7 100644
--- a/db/upgrade/native-sql-matrix.md
+++ b/db/upgrade/native-sql-matrix.md
@@ -82,6 +82,15 @@ of every production table. No gateway request or Docker e2e
coverage is claimed.
Per-engine artifacts contain flow results, durations and startup logs,
including
failed attempts. Cleanup is restricted to that job's generated test containers.
+Instance identity checks seed duplicate observations immediately before the
+2.7.2-to-2.7.3 migration. They require the newest heartbeat, update timestamp,
+creation timestamp and stable ID tie-break, retain different
namespaces/IPs/ports/
+types, and exercise maximum declared identity lengths. Both flows reject a real
+duplicate insert on `uk_instance_info_identity` and execute the actual mapper
+update SQL with stale, zero, newer and equal heartbeat timestamps. The H2
service
+suite additionally forces two independent writers to observe absence before
+either insert, in both insertion orders, and verifies caller-transaction
suspension.
+
## Local execution
With a Linux Docker daemon available, from the repository root:
diff --git a/db/upgrade/upgrade-guide.md b/db/upgrade/upgrade-guide.md
index 94d8249721..dbddf8da8a 100644
--- a/db/upgrade/upgrade-guide.md
+++ b/db/upgrade/upgrade-guide.md
@@ -27,6 +27,37 @@ Fresh installations create the renamed columns directly.
> this file is the Shenyu upgrade sql from v2.7.2 to v2.7.3
+### Instance identity uniqueness in 2.7.3
+
+The `2.7.2-upgrade-2.7.3-<dialect>.sql` scripts also enforce one
+`instance_info` row per `(namespace_id, instance_ip, instance_port,
instance_type)`.
+Stop **all** Admin writers and back up `instance_info` before running the
+one-time migration. Existing duplicate rows are deleted, retaining the complete
+row with the greatest heartbeat timestamp, then greatest `date_updated`,
+`date_created`, and finally `id`. Distinct identities and namespaces are
preserved;
+there are no references from gateway configuration to these diagnostic row IDs.
+Rows upgraded from 2.7.1 initially have heartbeat `0`, so timestamps and ID
decide
+the winner until a heartbeat has been persisted.
+
+Apply the schema migration before starting the updated Admin. Fresh
installations
+create `uk_instance_info_identity` directly. An existing persistent H2 database
+requires `2.7.2-upgrade-2.7.3-h2.sql`; rerunning `CREATE TABLE IF NOT EXISTS`
does
+not upgrade an existing table. H2 databases predating 2.7.2 must first have the
+`last_heart_beat_time BIGINT DEFAULT 0 NOT NULL` column added.
+
+The unique key uses all four columns without prefix truncation (434 characters,
+at most 1,736 bytes with MySQL utf8mb4). MySQL installations using a legacy
+767-byte InnoDB index limit must enable the modern large-prefix/DYNAMIC format
+or upgrade before applying it. Prefix indexes would conflate distinct
identities.
+
+Concurrent insert conflicts reload and update the canonical row. Updates accept
+only an equal or newer heartbeat; equal timestamps still allow liveness state
+changes without a fresh heartbeat. Zero or older heartbeats cannot overwrite a
+positive newer observation. This orders observations by their reported
timestamps;
+it does not establish a global clock or break ties between simultaneous states.
+Instance diagnostic writes run independently of caller transactions so a
duplicate
+insert does not leave PostgreSQL conflict recovery in an aborted transaction.
+
### Data permission uniqueness in 2.7.2
Stop Admin instances before running the `2.7.1-upgrade-2.7.2-<dialect>.sql`
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/InstanceInfoServiceImpl.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/InstanceInfoServiceImpl.java
index d3ff27c1a1..aba2fa4193 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/InstanceInfoServiceImpl.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/service/impl/InstanceInfoServiceImpl.java
@@ -29,7 +29,10 @@ import org.apache.shenyu.admin.utils.InstanceInfoUtils;
import org.apache.shenyu.common.utils.GsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
import java.sql.Timestamp;
import java.time.Instant;
@@ -51,6 +54,7 @@ public class InstanceInfoServiceImpl implements
InstanceInfoService {
}
@Override
+ @Transactional(propagation = Propagation.NOT_SUPPORTED)
public void createOrUpdate(final InstanceInfoVO instanceInfoVO) {
InstanceQuery instanceQuery = new InstanceQuery();
instanceQuery.setInstanceIp(instanceInfoVO.getInstanceIp());
@@ -63,10 +67,19 @@ public class InstanceInfoServiceImpl implements
InstanceInfoService {
InstanceInfoDO instanceInfoDO =
InstanceInfoDO.buildInstanceInfoDO(instanceInfoVO);
try {
instanceInfoMapper.insert(instanceInfoDO);
+ return;
+ } catch (DuplicateKeyException e) {
+ // Another Admin may have registered this identity after our
lookup.
+ // Each mapper call runs outside a transaction so PostgreSQL
can recover
+ // from the failed insert without leaving an aborted
transaction.
+ infoDO = instanceInfoMapper.selectOneByQuery(instanceQuery);
+ if (Objects.isNull(infoDO)) {
+ throw e;
+ }
} catch (Exception e) {
LOG.error("Failed to register instance info", e);
+ return;
}
- return;
}
LOG.debug("Update instance info: {}",
GsonUtils.getInstance().toJson(instanceInfoVO));
infoDO.setInstanceIp(instanceInfoVO.getInstanceIp());
diff --git a/shenyu-admin/src/main/resources/mappers/instance-info-sqlmap.xml
b/shenyu-admin/src/main/resources/mappers/instance-info-sqlmap.xml
index 61e4379f0a..32f8b1321d 100644
--- a/shenyu-admin/src/main/resources/mappers/instance-info-sqlmap.xml
+++ b/shenyu-admin/src/main/resources/mappers/instance-info-sqlmap.xml
@@ -59,6 +59,7 @@
last_heart_beat_time = #{lastHeartBeatTime, jdbcType=BIGINT}
WHERE
id = #{id, jdbcType=VARCHAR}
+ AND last_heart_beat_time <= #{lastHeartBeatTime, jdbcType=BIGINT}
</update>
<select id="findAllByNamespaceId" resultMap="BaseResultMap">
diff --git a/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
b/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
index 23a4bdb726..13dc381e2d 100644
--- a/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
+++ b/shenyu-admin/src/main/resources/sql-script/h2/schema.sql
@@ -1418,7 +1418,8 @@ CREATE TABLE IF NOT EXISTS instance_info
`date_created` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT
'date_created',
`date_updated` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP COMMENT 'date_updated',
`last_heart_beat_time` bigint NOT NULL DEFAULT 0 COMMENT 'last heart
beat time',
- PRIMARY KEY (`id`)
+ PRIMARY KEY (`id`),
+ CONSTRAINT uk_instance_info_identity UNIQUE (namespace_id, instance_ip,
instance_port, instance_type)
);
INSERT IGNORE INTO `permission` (`id`, `object_id`, `resource_id`,
`date_created`, `date_updated`) VALUES ('1697141926281382720',
'1346358560427216896', '1346778036402483300', '2023-08-31 06:59:01',
'2023-08-31 06:59:01');
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/mapper/InstanceInfoMapperTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/mapper/InstanceInfoMapperTest.java
index 9b0520854b..c35fb3e143 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/mapper/InstanceInfoMapperTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/mapper/InstanceInfoMapperTest.java
@@ -31,6 +31,7 @@ import jakarta.annotation.Resource;
import java.sql.Timestamp;
import java.util.List;
import java.util.Objects;
+import java.util.concurrent.atomic.AtomicInteger;
import static
org.apache.shenyu.common.constant.Constants.SYS_DEFAULT_NAMESPACE_ID;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -45,6 +46,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
public final class InstanceInfoMapperTest extends
AbstractSpringIntegrationTest {
+ private static final AtomicInteger NEXT_PORT = new AtomicInteger(10000);
+
private static final long FIXED_HEART_BEAT_TIME = 1700000000000L;
@Resource
@@ -198,12 +201,12 @@ public final class InstanceInfoMapperTest extends
AbstractSpringIntegrationTest
data.setDateCreated(now);
data.setDateUpdated(now);
data.setInstanceIp("127.0.0.1");
- data.setInstancePort("8080");
+ data.setInstancePort(Integer.toString(NEXT_PORT.getAndIncrement()));
data.setInstanceType("grpc");
data.setInstanceInfo("info");
data.setInstanceState(1);
data.setNamespaceId(SYS_DEFAULT_NAMESPACE_ID);
- data.setLastHeartBeatTime(System.currentTimeMillis());
+ data.setLastHeartBeatTime(FIXED_HEART_BEAT_TIME - 1);
return data;
}
}
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/mapper/InstanceInfoUpsertTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/mapper/InstanceInfoUpsertTest.java
new file mode 100644
index 0000000000..e001c9d568
--- /dev/null
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/mapper/InstanceInfoUpsertTest.java
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.admin.mapper;
+
+import org.apache.ibatis.builder.xml.XMLMapperBuilder;
+import org.apache.ibatis.io.Resources;
+import org.apache.ibatis.mapping.Environment;
+import org.apache.ibatis.session.Configuration;
+import org.apache.ibatis.session.SqlSessionFactoryBuilder;
+import org.apache.shenyu.admin.model.entity.InstanceInfoDO;
+import org.apache.shenyu.admin.model.query.InstanceQuery;
+import org.apache.shenyu.admin.model.vo.InstanceInfoVO;
+import org.apache.shenyu.admin.service.InstanceInfoService;
+import org.apache.shenyu.admin.service.impl.InstanceInfoServiceImpl;
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mybatis.spring.SqlSessionTemplate;
+import org.mybatis.spring.transaction.SpringManagedTransactionFactory;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.core.io.FileSystemResource;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import
org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Timestamp;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Exercises concurrent writers with real SQL sessions and the production H2
schema.
+ */
+public final class InstanceInfoUpsertTest {
+
+ private JdbcTemplate jdbc;
+
+ private InstanceInfoMapper mapper;
+
+ private DataSourceTransactionManager transactionManager;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ JdbcDataSource dataSource = new JdbcDataSource();
+ dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID() +
";MODE=MySQL;DB_CLOSE_DELAY=-1");
+ jdbc = new JdbcTemplate(dataSource);
+ transactionManager = new DataSourceTransactionManager(dataSource);
+ try (InputStream schema =
Resources.getResourceAsStream("sql-script/h2/schema.sql")) {
+ String sql = new String(schema.readAllBytes(),
StandardCharsets.UTF_8);
+ Matcher table = Pattern.compile("(?is)CREATE TABLE IF NOT EXISTS
instance_info\\s*\\(.*?\\);").matcher(sql);
+ assertTrue(table.find());
+ jdbc.execute(table.group());
+ }
+ Configuration configuration = new Configuration(new
Environment("test", new SpringManagedTransactionFactory(), dataSource));
+ configuration.setMapUnderscoreToCamelCase(true);
+ try (InputStream in =
Resources.getResourceAsStream("mappers/instance-info-sqlmap.xml")) {
+ new XMLMapperBuilder(in, configuration,
"mappers/instance-info-sqlmap.xml", configuration.getSqlFragments()).parse();
+ }
+ mapper = new SqlSessionTemplate(new
SqlSessionFactoryBuilder().build(configuration)).getMapper(InstanceInfoMapper.class);
+ }
+
+ @AfterEach
+ void tearDown() {
+ jdbc.execute("SHUTDOWN");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void concurrentRegistrationRetainsOneLatestObservation(final boolean
newerFirst) throws Exception {
+ CyclicBarrier absentLookups = new CyclicBarrier(2);
+ AtomicInteger lookups = new AtomicInteger();
+ CountDownLatch firstInsert = new CountDownLatch(1);
+ InstanceInfoMapper concurrentMapper = mock(InstanceInfoMapper.class,
delegatesTo(mapper));
+ doAnswer(invocation -> {
+ InstanceInfoDO found =
mapper.selectOneByQuery(invocation.getArgument(0));
+ if (lookups.incrementAndGet() <= 2) {
+ assertNull(found);
+ absentLookups.await(10, TimeUnit.SECONDS);
+ }
+ return found;
+ }).when(concurrentMapper).selectOneByQuery(any(InstanceQuery.class));
+ doAnswer(invocation -> {
+ InstanceInfoDO observation = invocation.getArgument(0);
+ boolean first = (observation.getLastHeartBeatTime() == 200) ==
newerFirst;
+ if (!first) {
+ assertTrue(firstInsert.await(10, TimeUnit.SECONDS));
+ }
+ try {
+ return mapper.insert(observation);
+ } finally {
+ if (first) {
+ firstInsert.countDown();
+ }
+ }
+ }).when(concurrentMapper).insert(any(InstanceInfoDO.class));
+ InstanceInfoServiceImpl service = new
InstanceInfoServiceImpl(concurrentMapper);
+ ExecutorService writers = Executors.newFixedThreadPool(2);
+ try {
+ Future<?> older = writers.submit(() ->
service.createOrUpdate(report("ns", 100, 2)));
+ Future<?> newer = writers.submit(() ->
service.createOrUpdate(report("ns", 200, 1)));
+ older.get(15, TimeUnit.SECONDS);
+ newer.get(15, TimeUnit.SECONDS);
+ List<InstanceInfoDO> rows = mapper.findAllByNamespaceId("ns");
+ assertEquals(1, rows.size(), "The database must enforce logical
identity across writers");
+ assertEquals(200, rows.get(0).getLastHeartBeatTime());
+ assertEquals(1, rows.get(0).getInstanceState());
+ assertEquals("info-200", rows.get(0).getInstanceInfo());
+ } finally {
+ writers.shutdownNow();
+ assertTrue(writers.awaitTermination(10, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ void
staleUpdateDoesNotOverwriteNewerReportAndNamespacesRemainIndependent() {
+ InstanceInfoServiceImpl service = new InstanceInfoServiceImpl(mapper);
+ service.createOrUpdate(report("ns", 200, 1));
+ final InstanceInfoDO initial =
mapper.findAllByNamespaceId("ns").get(0);
+ service.createOrUpdate(report("other", 50, 2));
+ service.createOrUpdate(report("ns", 100, 2));
+ service.createOrUpdate(report("ns", 0, 0));
+ InstanceInfoDO latest = mapper.findAllByNamespaceId("ns").get(0);
+ assertEquals(initial.getId(), latest.getId());
+ assertEquals(initial.getDateCreated(), latest.getDateCreated());
+ assertEquals(200, latest.getLastHeartBeatTime());
+ assertEquals(1, latest.getInstanceState());
+ assertEquals("info-200", latest.getInstanceInfo());
+ assertEquals(50,
mapper.findAllByNamespaceId("other").get(0).getLastHeartBeatTime());
+
+ // Liveness can transition to offline without a new heartbeat.
+ service.createOrUpdate(report("ns", 200, 2));
+ assertEquals(2,
mapper.findAllByNamespaceId("ns").get(0).getInstanceState());
+ service.createOrUpdate(report("ns", 300, 1));
+ assertEquals(300,
mapper.findAllByNamespaceId("ns").get(0).getLastHeartBeatTime());
+ assertEquals(2, mapper.selectAll().size());
+ }
+
+ @Test
+ void conflictRecoverySuspendsTheCallersTransaction() {
+ new InstanceInfoServiceImpl(mapper).createOrUpdate(report("ns", 100,
2));
+ AtomicInteger lookups = new AtomicInteger();
+ InstanceInfoMapper conflictingMapper = mock(InstanceInfoMapper.class,
delegatesTo(mapper));
+ doAnswer(invocation -> lookups.incrementAndGet() == 1 ? null :
mapper.selectOneByQuery(invocation.getArgument(0)))
+
.when(conflictingMapper).selectOneByQuery(any(InstanceQuery.class));
+ ProxyFactory factory = new ProxyFactory(new
InstanceInfoServiceImpl(conflictingMapper));
+ factory.addAdvice(new TransactionInterceptor(transactionManager, new
AnnotationTransactionAttributeSource()));
+ InstanceInfoService service = (InstanceInfoService) factory.getProxy();
+ new
TransactionTemplate(transactionManager).executeWithoutResult(status -> {
+ service.createOrUpdate(report("ns", 200, 1));
+ assertEquals(1, jdbc.queryForObject("SELECT COUNT(*) FROM
instance_info", Integer.class));
+ status.setRollbackOnly();
+ });
+ assertEquals(200,
mapper.findAllByNamespaceId("ns").get(0).getLastHeartBeatTime());
+ }
+
+ @Test
+ void migrationRetainsLatestCompleteRowsAndRejectsDuplicates() {
+ jdbc.execute("ALTER TABLE instance_info DROP CONSTRAINT
uk_instance_info_identity");
+ jdbc.execute("CREATE TABLE selector (loged INT)");
+ jdbc.execute("CREATE TABLE rule (loged INT)");
+ insertLegacy("old-heartbeat", "ns", 100, "2026-01-03 00:00:00");
+ insertLegacy("latest-heartbeat", "ns", 200, "2026-01-01 00:00:00");
+ insertLegacy("tie-a", "ns-tie", 0, "2026-01-01 00:00:00");
+ insertLegacy("tie-z", "ns-tie", 0, "2026-01-01 00:00:00");
+ insertLegacy("other", "ns-other", 100, "2026-01-01 00:00:00");
+ Path root = Path.of(System.getProperty("user.dir")).toAbsolutePath();
+ while (Objects.nonNull(root) &&
!Files.isDirectory(root.resolve("db/upgrade"))) {
+ root = root.getParent();
+ }
+ assertNotNull(root);
+ new ResourceDatabasePopulator(new
FileSystemResource(root.resolve("db/upgrade/2.7.2-upgrade-2.7.3-h2.sql")))
+ .execute(Objects.requireNonNull(jdbc.getDataSource()));
+ assertEquals(Set.of("latest-heartbeat", "tie-z", "other"),
mapper.selectAll().stream().map(InstanceInfoDO::getId).collect(Collectors.toSet()));
+ InstanceInfoDO retained = mapper.selectById("latest-heartbeat");
+ assertEquals(200, retained.getLastHeartBeatTime());
+ assertEquals("info-200", retained.getInstanceInfo());
+ assertEquals(Timestamp.valueOf("2026-01-01 00:00:00"),
retained.getDateUpdated());
+ assertThrows(DuplicateKeyException.class, () ->
insertLegacy("conflict", "ns", 300, "2026-01-04 00:00:00"));
+ }
+
+ private void insertLegacy(final String id, final String namespace, final
long heartbeat, final String updated) {
+ InstanceInfoDO row =
InstanceInfoDO.buildInstanceInfoDO(report(namespace, heartbeat, 1));
+ row.setId(id);
+ row.setDateCreated(Timestamp.valueOf("2026-01-01 00:00:00"));
+ row.setDateUpdated(Timestamp.valueOf(updated));
+ mapper.insert(row);
+ }
+
+ private InstanceInfoVO report(final String namespace, final long
heartbeat, final int state) {
+ InstanceInfoVO report = new InstanceInfoVO();
+ report.setNamespaceId(namespace);
+ report.setInstanceIp("127.0.0.1");
+ report.setInstancePort("9195");
+ report.setInstanceType("bootstrap");
+ report.setInstanceInfo("info-" + heartbeat);
+ report.setLastHeartBeatTime(heartbeat);
+ report.setInstanceState(state);
+ return report;
+ }
+}
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/impl/InstanceInfoServiceImplTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/impl/InstanceInfoServiceImplTest.java
index 1bc42618ed..693ea84e6c 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/impl/InstanceInfoServiceImplTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/service/impl/InstanceInfoServiceImplTest.java
@@ -31,15 +31,18 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import org.springframework.dao.DuplicateKeyException;
import java.util.Arrays;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.never;
/**
* Test case for {@link InstanceInfoServiceImpl}.
@@ -96,6 +99,25 @@ public final class InstanceInfoServiceImplTest {
assertEquals(FIXED_HEART_BEAT_TIME,
captor.getValue().getLastHeartBeatTime());
}
+ @Test
+ public void testDuplicateRegistrationUpdatesTheCanonicalRow() {
+
given(instanceInfoMapper.selectOneByQuery(any(InstanceQuery.class))).willReturn(null,
buildDO("winner"));
+
given(instanceInfoMapper.insert(any(InstanceInfoDO.class))).willThrow(new
DuplicateKeyException("identity conflict"));
+ instanceInfoService.createOrUpdate(buildReport());
+ ArgumentCaptor<InstanceInfoDO> captor =
ArgumentCaptor.forClass(InstanceInfoDO.class);
+ verify(instanceInfoMapper).updateById(captor.capture());
+ assertEquals("winner", captor.getValue().getId());
+ assertEquals(FIXED_HEART_BEAT_TIME,
captor.getValue().getLastHeartBeatTime());
+ }
+
+ @Test
+ public void
testUnrelatedPrimaryKeyConflictIsNotTreatedAsAnIdentityConflict() {
+
given(instanceInfoMapper.selectOneByQuery(any(InstanceQuery.class))).willReturn(null);
+
given(instanceInfoMapper.insert(any(InstanceInfoDO.class))).willThrow(new
DuplicateKeyException("primary key conflict"));
+ assertThrows(DuplicateKeyException.class, () ->
instanceInfoService.createOrUpdate(buildReport()));
+ verify(instanceInfoMapper, never()).updateById(any());
+ }
+
private InstanceInfoVO buildReport() {
InstanceInfoVO report = new InstanceInfoVO();
report.setInstanceIp("127.0.0.1");
