@@ -49,7 +49,8 @@ ovs_pytests = \
python/ovs/tests/test_kv.py \
python/ovs/tests/test_list.py \
python/ovs/tests/test_odp.py \
- python/ovs/tests/test_ofp.py
+ python/ovs/tests/test_ofp.py \
+ python/ovs/tests/test_reconciled_notices.py
ovs_flowviz = \
python/ovs/flowviz/__init__.py \
@@ -47,6 +47,66 @@ Notice = collections.namedtuple('Notice', ('event', 'row', 'updates'))
Notice.__new__.__defaults__ = (None,) # default updates=None
+class ReconciledNotices:
+ """Lazy view over pending notices, reconciling reconnect events.
+
+ On reconnect the IDL discards its local replica with __clear() and
+ repopulates it from a fresh monitor dump. When notify_reconnect()
+ is in use, __clear() records a ROW_DELETE notice for every
+ pre-existing row and the subsequent dump records a ROW_CREATE notice
+ for every row in the new snapshot. This view reconciles the
+ DELETE+CREATE pair collected per UUID:
+
+ - Same data: suppressed (the row survived the reconnect unchanged)
+ - Different data: emitted as ROW_UPDATE with only the changed columns
+ - DELETE only: emitted (row was deleted while disconnected)
+ - CREATE only: emitted (row appeared while disconnected)
+
+ Iteration yields the reconciled Notice objects. __getitem__(uuid)
+ returns the reconciled Notice for a specific UUID, or None if that
+ UUID reconciles away. Reconciliation is computed on access; nothing
+ is cached.
+ """
+
+ def __init__(self, notices):
+ self._raw = notices
+
+ def __getitem__(self, row_uuid):
+ return self._reconcile(row_uuid, self._raw[row_uuid])
+
+ def __iter__(self):
+ for row_uuid in self._raw:
+ notice = self._reconcile(row_uuid, self._raw[row_uuid])
+ if notice is not None:
+ yield notice
+
+ @staticmethod
+ def _reconcile(row_uuid, events):
+ if len(events) == 1:
+ return events[0]
+ if (
+ len(events) == 2 and
+ events[0].event == ROW_DELETE and
+ events[1].event == ROW_CREATE
+ ):
+ old_row = events[0].row
+ new_row = events[1].row
+ old_data = {}
+ for col in old_row._table.columns:
+ if col in old_row._data and col in new_row._data:
+ if old_row._data[col] != new_row._data[col]:
+ old_data[col] = old_row._data[col]
+ if old_data:
+ # Carry the real idl reference (from the new row) on the
+ # "updates" row so reads of reference-typed columns can
+ # resolve, matching notify()'s ROW_UPDATE semantics.
+ return Notice(ROW_UPDATE, new_row,
+ Row(new_row._idl, old_row._table, row_uuid,
+ old_data))
+ return None
+ assert False, "unexpected number of events for row %s" % row_uuid
+
+
class ColumnDefaultDict(dict):
"""A column dictionary with on-demand generated default values
@@ -315,6 +375,12 @@ class Idl(object):
self.cond_changed = False
self.cond_seqno = 0
+ # Notices produced while parsing a single monitor reply (and the
+ # DELETE notices produced by __clear() on reconnect) are collected
+ # here keyed by row UUID, then flushed once parsing is complete.
+ # See _flush_notices().
+ self._pending_notices = collections.defaultdict(list)
+
def _parse_remotes(self, remote):
# If remote is -
# "tcp:10.0.0.1:6641,unix:/tmp/db.sock,t,s,tcp:10.0.0.2:6642"
@@ -487,6 +553,11 @@ class Idl(object):
and self._monitor_request_id == msg.id):
# Reply to our "monitor" request.
try:
+ # A non-zero change_seqno means we have downloaded the
+ # database before, so this reply repopulates the replica
+ # after a reconnect rather than being the initial
+ # download. Compute this before the increment below.
+ reconnect = self.has_ever_connected()
self.change_seqno += 1
self._monitor_request_id = None
if (self.state ==
@@ -494,15 +565,23 @@ class Idl(object):
# If 'found' is false, clear table rows for new dump
if not msg.result[0]:
self.__clear()
- self.__parse_update(msg.result[2], OVSDB_UPDATE3)
+ self.__parse_update(msg.result[2], OVSDB_UPDATE3,
+ reconnect=reconnect)
+ else:
+ # 'found' is true: the server sent an incremental
+ # update since last_id, already reconciled, so
+ # deliver it through notify() as usual.
+ self.__parse_update(msg.result[2], OVSDB_UPDATE3)
self.last_id = msg.result[1]
elif self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED:
self.__clear()
- self.__parse_update(msg.result, OVSDB_UPDATE2)
+ self.__parse_update(msg.result, OVSDB_UPDATE2,
+ reconnect=reconnect)
else:
assert self.state == self.IDL_S_DATA_MONITOR_REQUESTED
self.__clear()
- self.__parse_update(msg.result, OVSDB_UPDATE)
+ self.__parse_update(msg.result, OVSDB_UPDATE,
+ reconnect=reconnect)
self.state = self.IDL_S_MONITORING
except error.Error as e:
@@ -773,6 +852,11 @@ class Idl(object):
def notify(self, event, row, updates=None):
"""Hook for implementing create/update/delete notifications
+ This is called once per changed row for incremental updates and for
+ the initial database download. On reconnect the changes are instead
+ delivered to notify_reconnect(), so that the redundant DELETE+CREATE
+ churn from rebuilding the local replica can be reconciled first.
+
:param event: The event that was triggered
:type event: ROW_CREATE, ROW_UPDATE, or ROW_DELETE
:param row: The row as it is after the operation has occured
@@ -782,6 +866,40 @@ class Idl(object):
:type updates: Row
"""
+ def notify_reconnect(self, notices):
+ """Hook for delivering reconciled changes after a reconnect
+
+ On reconnect the IDL discards its local replica and repopulates it
+ from a fresh monitor dump, which would otherwise appear as a
+ DELETE+CREATE for every row. Rather than replaying that churn, the
+ changes that actually occurred while the IDL was disconnected are
+ collected, reconciled, and delivered through this hook.
+
+ The initial database download and all incremental updates still flow
+ through notify() as usual; this hook is only used for the
+ post-reconnect monitor dump.
+
+ The default implementation forwards each reconciled notice to
+ notify(), so that a subclass which only overrides notify() still
+ receives post-reconnect changes -- reconciled (rows that did not
+ change are suppressed, modified rows arrive as a single ROW_UPDATE)
+ rather than as a create-for-every-row storm. Override this hook to
+ receive the whole reconciled set in one call instead; doing so
+ suppresses the per-row notify() calls for the reconnect dump.
+
+ :param notices: A ReconciledNotices view. Iterating it yields one
+ Notice per row that actually changed while the IDL
+ was disconnected: a ROW_UPDATE (with only the changed
+ columns in its 'updates') for a modified row, a
+ ROW_DELETE for a row that went away, or a ROW_CREATE
+ for a row that appeared. Rows that survived the
+ reconnect unchanged are suppressed. The view can
+ also be indexed by row UUID.
+ :type notices: ReconciledNotices
+ """
+ for notice in notices:
+ self.notify(*notice)
+
def cooperative_yield(self):
"""Hook for cooperatively yielding to eventlet/gevent/asyncio/etc.
@@ -797,6 +915,14 @@ class Idl(object):
for table in self.tables.values():
if table.rows:
changed = True
+ # Record a DELETE notice for every row being discarded. The
+ # subsequent monitor dump records a matching CREATE for rows
+ # that still exist; _flush_notices() reconciles the pairs on
+ # reconnect. On a first connect the tables are empty, so this
+ # records nothing.
+ for row_uuid, row in table.rows.items():
+ self._pending_notices[row_uuid].append(
+ Notice(ROW_DELETE, row))
table.rows.clear()
self.cond_seqno = 0
@@ -806,12 +932,16 @@ class Idl(object):
def __update_has_lock(self, new_has_lock):
if new_has_lock and not self.has_lock:
- if self._monitor_request_id is None:
+ if self.state == self.IDL_S_MONITORING:
self.change_seqno += 1
else:
- # We're waiting for a monitor reply, so don't signal that the
- # database changed. The monitor reply will increment
- # change_seqno anyhow.
+ # We haven't finished downloading the database yet, so don't
+ # signal that the database changed. The monitor reply will
+ # increment change_seqno anyhow. Gating on the MONITORING
+ # state (rather than merely on no monitor request being in
+ # flight) keeps change_seqno at 0 through the initial
+ # download, so that has_ever_connected() reliably
+ # distinguishes a first connect from a reconnect.
pass
self.is_lock_contended = False
self.has_lock = new_has_lock
@@ -919,7 +1049,7 @@ class Idl(object):
self._server_monitor_request_id = msg.id
self.send_request(msg)
- def __parse_update(self, update, version, tables=None):
+ def __parse_update(self, update, version, tables=None, reconnect=False):
try:
if not tables:
self.__do_parse_update(update, version, self.tables)
@@ -928,13 +1058,38 @@ class Idl(object):
except error.Error as e:
vlog.err("%s: error parsing update: %s"
% (self._session.get_name(), e))
+ # Drop any notices buffered before the error rather than
+ # delivering a partial, inconsistent update.
+ self._pending_notices.clear()
+ return
+ self._flush_notices(reconnect)
+
+ def _flush_notices(self, reconnect):
+ """Deliver the notices buffered while parsing a monitor reply.
+
+ On reconnect the buffered per-row DELETE (from __clear()) and CREATE
+ events are reconciled and handed to notify_reconnect() in a single
+ call. Otherwise (initial download and incremental updates) each
+ buffered notice is delivered individually through notify().
+ """
+ if reconnect:
+ # Hand the buffered notices to the view and start a fresh buffer,
+ # so the (lazy) ReconciledNotices remains valid even if the
+ # callback holds on to it past this call.
+ pending = self._pending_notices
+ self._pending_notices = collections.defaultdict(list)
+ self.notify_reconnect(ReconciledNotices(pending))
+ else:
+ for events in self._pending_notices.values():
+ for notice in events:
+ self.notify(*notice)
+ self._pending_notices.clear()
def __do_parse_update(self, table_updates, version, tables):
if not isinstance(table_updates, dict):
raise error.Error("<table-updates> is not an object",
table_updates)
- notices = []
for table_name, table_update in table_updates.items():
table = tables.get(table_name)
if not table:
@@ -964,7 +1119,7 @@ class Idl(object):
if version in (OVSDB_UPDATE2, OVSDB_UPDATE3):
changes = self.__process_update2(table, uuid, row_update)
if changes and tables is not self.server_tables:
- notices.append(changes)
+ self._pending_notices[uuid].append(changes)
self.change_seqno += 1
continue
@@ -979,10 +1134,8 @@ class Idl(object):
changes = self.__process_update(table, uuid, old, new)
if changes and tables is not self.server_tables:
- notices.append(changes)
+ self._pending_notices[uuid].append(changes)
self.change_seqno += 1
- for notice in notices:
- self.notify(*notice)
def __process_update2(self, table, uuid, row_update):
"""Returns Notice if a column changed, False otherwise."""
new file mode 100644
@@ -0,0 +1,255 @@
+import collections
+import uuid
+
+import pytest
+
+from ovs.db.idl import (
+ Notice,
+ ReconciledNotices,
+ ROW_CREATE,
+ ROW_DELETE,
+ ROW_UPDATE,
+ Row,
+)
+
+
+class FakeTable:
+ def __init__(self, column_names):
+ self.columns = {name: None for name in column_names}
+
+
+class FakeDatum:
+ """Minimal Datum-like object that supports equality comparison."""
+ def __init__(self, value):
+ self.value = value
+
+ def __eq__(self, other):
+ if isinstance(other, FakeDatum):
+ return self.value == other.value
+ return NotImplemented
+
+ def __repr__(self):
+ return "FakeDatum(%r)" % self.value
+
+
+def make_row(table, row_uuid, data):
+ return Row(None, table, row_uuid, data)
+
+
+def make_notices(notice_list):
+ """Build defaultdict(list) keyed by UUID from (uuid, Notice) pairs."""
+ raw = collections.defaultdict(list)
+ for row_uuid, notice in notice_list:
+ raw[row_uuid].append(notice)
+ return raw
+
+
+TABLE = FakeTable(["name", "value", "tag"])
+
+
+@pytest.mark.parametrize(
+ "event",
+ [ROW_CREATE, ROW_DELETE, ROW_UPDATE],
+)
+def test_single_event_passthrough(event):
+ row_uuid = uuid.uuid4()
+ row = make_row(TABLE, row_uuid, {"name": FakeDatum("r1")})
+ if event == ROW_UPDATE:
+ updates = make_row(TABLE, row_uuid, {"name": FakeDatum("old")})
+ notice = Notice(event, row, updates)
+ else:
+ notice = Notice(event, row)
+ raw = make_notices([(row_uuid, notice)])
+
+ result = list(ReconciledNotices(raw))
+
+ assert len(result) == 1
+ assert result[0] is notice
+
+
+@pytest.mark.parametrize(
+ "old_data,new_data",
+ [
+ (
+ {"name": FakeDatum("same")},
+ {"name": FakeDatum("same")},
+ ),
+ (
+ {"name": FakeDatum("n"), "value": FakeDatum(42),
+ "tag": FakeDatum("t")},
+ {"name": FakeDatum("n"), "value": FakeDatum(42),
+ "tag": FakeDatum("t")},
+ ),
+ ],
+)
+def test_delete_create_unchanged_suppressed(old_data, new_data):
+ row_uuid = uuid.uuid4()
+ raw = make_notices([
+ (row_uuid, Notice(ROW_DELETE,
+ make_row(TABLE, row_uuid, old_data))),
+ (row_uuid, Notice(ROW_CREATE,
+ make_row(TABLE, row_uuid, new_data))),
+ ])
+
+ result = list(ReconciledNotices(raw))
+
+ assert len(result) == 0
+
+
+def test_delete_create_unchanged_getitem_returns_none():
+ row_uuid = uuid.uuid4()
+ data = {"name": FakeDatum("same")}
+ raw = make_notices([
+ (row_uuid, Notice(ROW_DELETE,
+ make_row(TABLE, row_uuid, dict(data)))),
+ (row_uuid, Notice(ROW_CREATE,
+ make_row(TABLE, row_uuid, dict(data)))),
+ ])
+
+ assert ReconciledNotices(raw)[row_uuid] is None
+
+
+@pytest.mark.parametrize(
+ "old_data,new_data,expected_changed_cols",
+ [
+ (
+ {"name": FakeDatum("old_val")},
+ {"name": FakeDatum("new_val")},
+ {"name"},
+ ),
+ (
+ {"name": FakeDatum("n"), "value": FakeDatum(1),
+ "tag": FakeDatum("t")},
+ {"name": FakeDatum("n"), "value": FakeDatum(2),
+ "tag": FakeDatum("t")},
+ {"value"},
+ ),
+ (
+ {"name": FakeDatum("a"), "value": FakeDatum(1)},
+ {"name": FakeDatum("b"), "value": FakeDatum(2)},
+ {"name", "value"},
+ ),
+ ],
+)
+def test_delete_create_changed_becomes_update(old_data, new_data,
+ expected_changed_cols):
+ row_uuid = uuid.uuid4()
+ old_row = make_row(TABLE, row_uuid, old_data)
+ new_row = make_row(TABLE, row_uuid, new_data)
+ raw = make_notices([
+ (row_uuid, Notice(ROW_DELETE, old_row)),
+ (row_uuid, Notice(ROW_CREATE, new_row)),
+ ])
+
+ result = list(ReconciledNotices(raw))
+
+ assert len(result) == 1
+ notice = result[0]
+ assert notice.event == ROW_UPDATE
+ assert notice.row is new_row
+ assert set(notice.updates._data.keys()) == expected_changed_cols
+ for col in expected_changed_cols:
+ assert notice.updates._data[col] == old_data[col]
+
+
+def test_update_row_has_correct_uuid_and_table():
+ row_uuid = uuid.uuid4()
+ old_row = make_row(TABLE, row_uuid, {"name": FakeDatum("old")})
+ new_row = make_row(TABLE, row_uuid, {"name": FakeDatum("new")})
+ raw = make_notices([
+ (row_uuid, Notice(ROW_DELETE, old_row)),
+ (row_uuid, Notice(ROW_CREATE, new_row)),
+ ])
+
+ result = list(ReconciledNotices(raw))
+
+ assert result[0].updates.uuid == row_uuid
+ assert result[0].updates._table is TABLE
+
+
+@pytest.mark.parametrize(
+ "old_data,new_data",
+ [
+ (
+ {"name": FakeDatum("n"), "value": FakeDatum("x")},
+ {"name": FakeDatum("n")},
+ ),
+ (
+ {"name": FakeDatum("n")},
+ {"name": FakeDatum("n"), "value": FakeDatum("x")},
+ ),
+ ],
+)
+def test_column_only_in_one_row_not_compared(old_data, new_data):
+ row_uuid = uuid.uuid4()
+ raw = make_notices([
+ (row_uuid, Notice(ROW_DELETE,
+ make_row(TABLE, row_uuid, old_data))),
+ (row_uuid, Notice(ROW_CREATE,
+ make_row(TABLE, row_uuid, new_data))),
+ ])
+
+ result = list(ReconciledNotices(raw))
+
+ assert len(result) == 0
+
+
+def test_mixed_rows():
+ uuid_new = uuid.uuid4()
+ uuid_del = uuid.uuid4()
+ uuid_unchanged = uuid.uuid4()
+ uuid_changed = uuid.uuid4()
+
+ row_new = make_row(TABLE, uuid_new, {"name": FakeDatum("new_row")})
+ row_del = make_row(TABLE, uuid_del, {"name": FakeDatum("gone")})
+ row_unch_old = make_row(TABLE, uuid_unchanged, {"name": FakeDatum("s")})
+ row_unch_new = make_row(TABLE, uuid_unchanged, {"name": FakeDatum("s")})
+ row_chg_old = make_row(TABLE, uuid_changed, {"name": FakeDatum("before")})
+ row_chg_new = make_row(TABLE, uuid_changed, {"name": FakeDatum("after")})
+
+ raw = make_notices([
+ (uuid_new, Notice(ROW_CREATE, row_new)),
+ (uuid_del, Notice(ROW_DELETE, row_del)),
+ (uuid_unchanged, Notice(ROW_DELETE, row_unch_old)),
+ (uuid_unchanged, Notice(ROW_CREATE, row_unch_new)),
+ (uuid_changed, Notice(ROW_DELETE, row_chg_old)),
+ (uuid_changed, Notice(ROW_CREATE, row_chg_new)),
+ ])
+
+ results = {n.row.uuid: n for n in ReconciledNotices(raw)}
+
+ assert len(results) == 3
+ assert results[uuid_new].event == ROW_CREATE
+ assert results[uuid_del].event == ROW_DELETE
+ assert results[uuid_changed].event == ROW_UPDATE
+ assert uuid_unchanged not in results
+
+
+def test_all_rows_suppressed():
+ uuids = [uuid.uuid4() for _ in range(3)]
+ raw = make_notices(
+ [(u, Notice(ROW_DELETE, make_row(TABLE, u, {"name": FakeDatum("x")})))
+ for u in uuids]
+ + [(u, Notice(ROW_CREATE,
+ make_row(TABLE, u, {"name": FakeDatum("x")})))
+ for u in uuids]
+ )
+
+ assert list(ReconciledNotices(raw)) == []
+
+
+def test_empty_notices():
+ assert list(ReconciledNotices(collections.defaultdict(list))) == []
+
+
+def test_unexpected_event_count_asserts():
+ row_uuid = uuid.uuid4()
+ row = make_row(TABLE, row_uuid, {"name": FakeDatum("x")})
+ raw = make_notices([
+ (row_uuid, Notice(ROW_DELETE, row)),
+ (row_uuid, Notice(ROW_CREATE, row)),
+ (row_uuid, Notice(ROW_UPDATE, row)),
+ ])
+
+ with pytest.raises(AssertionError, match="unexpected number of events"):
+ list(ReconciledNotices(raw))
@@ -2099,13 +2099,32 @@ OVSDB_CHECK_IDL_NOTIFY([simple idl verify notify],
012: table simple: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
012: table simple: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
013: reconnect
-014: event:create, row={i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[]}, uuid=<6>, updates=None
-014: event:create, row={i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>]}, uuid=<0>, updates=None
014: table simple: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
014: table simple: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
015: done
]])
+dnl When a client overrides notify_reconnect(), the post-reconnect monitor
+dnl dump is delivered there (reconciled) instead of as per-row notify()
+dnl calls. A row that is unchanged across the reconnect reconciles away, so
+dnl notify_reconnect() receives an empty set and notify() is not called.
+OVSDB_CHECK_IDL_PY([simple idl, notify_reconnect override], [],
+ [['track-notify-reconnect' \
+ '["idltest",
+ {"op": "insert",
+ "table": "simple",
+ "row": {"i": 1, "s": "mystring"}}]' \
+ 'reconnect']],
+ [[000: empty
+001: {"error":null,"result":[{"uuid":["uuid","<0>"]}]}
+002: event:create, row={i=1 r=0 b=false s=mystring u=<1> ia=[] ra=[] ba=[] sa=[] ua=[]}, uuid=<0>, updates=None
+002: table simple: i=1 r=0 b=false s=mystring u=<1> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<0>
+003: reconnect
+004: notify_reconnect:
+004: table simple: i=1 r=0 b=false s=mystring u=<1> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<0>
+005: done
+]], [notify])
+
OVSDB_CHECK_IDL_NOTIFY([indexed idl, modification and removal notify],
[['track-notify' \
'["idltest",
@@ -758,6 +758,7 @@ def update_condition(idl, commands, step):
def do_idl(schema_file, remote, *commands):
schema_helper = ovs.db.idl.SchemaHelper(schema_file)
track_notify = False
+ track_notify_reconnect = False
if remote.startswith("ssl:"):
if len(commands) < 3:
@@ -774,6 +775,11 @@ def do_idl(schema_file, remote, *commands):
commands = commands[1:]
track_notify = True
+ if commands and commands[0] == "track-notify-reconnect":
+ commands = commands[1:]
+ track_notify = True
+ track_notify_reconnect = True
+
if commands and commands[0].startswith("?"):
readonly = {}
for x in commands[0][1:].split("?"):
@@ -831,8 +837,26 @@ def do_idl(schema_file, remote, *commands):
sys.stdout.write(output)
sys.stdout.flush()
+ def mock_notify_reconnect(notices):
+ output = "%03d: notify_reconnect:\n" % step
+ for notice in notices:
+ output += "%03d: event:%s, row={%s}, %s, updates=" % (
+ step, str(notice.event),
+ get_simple_table_printable_row(notice.row, 'l2', 'l1'),
+ get_simple_printable_row_string(notice.row, ["uuid"]))
+ if notice.updates is None:
+ output += "None"
+ else:
+ output += "{" + get_simple_table_printable_row(
+ notice.updates) + "}"
+ output += "\n"
+ sys.stdout.write(output)
+ sys.stdout.flush()
+
if track_notify and "simple" in idl.tables:
idl.notify = mock_notify
+ if track_notify_reconnect:
+ idl.notify_reconnect = mock_notify_reconnect
commands = list(commands)
if len(commands) >= 1 and "condition" in commands[0]:
On reconnect the Python IDL discards its local replica with __clear() and repopulates it from a fresh monitor dump. __clear() dropped the old rows silently, so clients saw only a ROW_CREATE for every row in the new snapshot -- a create-for-every-row "storm" that told the client nothing about what actually changed while it was disconnected (rows that were unchanged or modified looked identical: a fresh create, deletes missing). Record a ROW_DELETE for each row __clear() discards, collect the notices produced while parsing a monitor reply (keyed by row UUID) instead of calling notify() inline, and on reconnect deliver them through a new notify_reconnect() hook. The notices are handed over as a ReconciledNotices view that pairs the __clear() DELETE with the re-dump CREATE per UUID and reconciles them: - unchanged row: suppressed - modified row: single ROW_UPDATE with only changed columns - row deleted while gone: ROW_DELETE - row created while gone: ROW_CREATE The initial database download and all incremental updates still flow through notify() unchanged. The default notify_reconnect() forwards each reconciled notice to notify(), so a subclass that only overrides notify() still learns about post-reconnect changes (now reconciled rather than as a storm); overriding notify_reconnect() receives the whole set in one call and suppresses the per-row notify() calls for the reconnect dump. First connect is distinguished from reconnect via has_ever_connected(). To keep that signal honest, only bump change_seqno on a lock reply once the IDL has reached the MONITORING state, so lock acquisition during the initial handshake no longer makes a first connect look like a reconnect. Adds pytest coverage for ReconciledNotices and functional coverage for both the default (notify()) and overridden notify_reconnect() paths. Assisted-by: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: Terry Wilson <twilson@redhat.com> --- v3: Note: Supersedes python: Add notify_on_clear for reconciled notices. - Removes opt-in for the main patch - Adds notify_reconnect() instead of batch_notify() python/automake.mk | 3 +- python/ovs/db/idl.py | 179 +++++++++++++- python/ovs/tests/test_reconciled_notices.py | 255 ++++++++++++++++++++ tests/ovsdb-idl.at | 23 +- tests/test-ovsdb.py | 24 ++ 5 files changed, 468 insertions(+), 16 deletions(-) create mode 100644 python/ovs/tests/test_reconciled_notices.py