diff --git a/lib/ovsdb-cs.c b/lib/ovsdb-cs.c
index f37aa5b04414..70fe10bb33f6 100644
--- a/lib/ovsdb-cs.c
+++ b/lib/ovsdb-cs.c
@@ -39,6 +39,1945 @@
 #include "uuid.h"
 
 VLOG_DEFINE_THIS_MODULE(ovsdb_cs);
+
+/* Connection state machine.
+ *
+ * When a JSON-RPC session connects, the CS layer sends a "monitor_cond"
+ * request for the Database table in the _Server database and transitions to
+ * the CS_S_SERVER_MONITOR_REQUESTED state.  If the session drops and
+ * reconnects, or if the CS receives a "monitor_canceled" notification for a
+ * table it is monitoring, the CS starts over again in the same way. */
+#define OVSDB_CS_STATES                                                 \
+    /* Waits for "get_schema" reply, then sends "monitor_cond"          \
+     * request for the Database table in the _Server database, whose    \
+     * details are informed by the schema, and transitions to           \
+     * CS_S_SERVER_MONITOR_REQUESTED. */                                \
+    OVSDB_CS_STATE(SERVER_SCHEMA_REQUESTED)                             \
+                                                                        \
+    /* Waits for "monitor_cond" reply for the Database table:           \
+     *                                                                  \
+     * - If the reply indicates success, and the Database table has a   \
+     *   row for the CS database:                                       \
+     *                                                                  \
+     *   * If the row indicates that this is a clustered database       \
+     *     that is not connected to the cluster, closes the             \
+     *     connection.  The next connection attempt has a chance at     \
+     *     picking a connected server.                                  \
+     *                                                                  \
+     *   * Otherwise, sends a monitoring request for the CS             \
+     *     database whose details are informed by the schema            \
+     *     (obtained from the row), and transitions to                  \
+     *     CS_S_DATA_MONITOR_(COND_(SINCE_))REQUESTED.                  \
+     *                                                                  \
+     * - If the reply indicates success, but the Database table does    \
+     *   not have a row for the CS database, transitions to             \
+     *   CS_S_ERROR.                                                    \
+     *                                                                  \
+     * - If the reply indicates failure, sends a "get_schema" request   \
+     *   for the CS database and transitions to                         \
+     *   CS_S_DATA_SCHEMA_REQUESTED. */                                 \
+    OVSDB_CS_STATE(SERVER_MONITOR_REQUESTED)                            \
+                                                                        \
+    /* Waits for "get_schema" reply, then sends "monitor_cond"          \
+     * request whose details are informed by the schema, and            \
+     * transitions to CS_S_DATA_MONITOR_COND_REQUESTED. */              \
+    OVSDB_CS_STATE(DATA_SCHEMA_REQUESTED)                               \
+                                                                        \
+    /* Waits for "monitor_cond_since" reply.  If successful, replaces   \
+     * the CS contents by the data carried in the reply and             \
+     * transitions to CS_S_MONITORING.  On failure, sends a             \
+     * "monitor_cond" request and transitions to                        \
+     * CS_S_DATA_MONITOR_COND_REQUESTED. */                             \
+    OVSDB_CS_STATE(DATA_MONITOR_COND_SINCE_REQUESTED)                   \
+                                                                        \
+    /* Waits for "monitor_cond" reply.  If successful, replaces the     \
+     * CS contents by the data carried in the reply and transitions     \
+     * to CS_S_MONITORING.  On failure, sends a "monitor" request       \
+     * and transitions to CS_S_DATA_MONITOR_REQUESTED. */               \
+    OVSDB_CS_STATE(DATA_MONITOR_COND_REQUESTED)                         \
+                                                                        \
+    /* Waits for "monitor" reply.  If successful, replaces the CS       \
+     * contents by the data carried in the reply and transitions to     \
+     * CS_S_MONITORING.  On failure, transitions to CS_S_ERROR. */      \
+    OVSDB_CS_STATE(DATA_MONITOR_REQUESTED)                              \
+                                                                        \
+    /* State that processes "update", "update2" or "update3"            \
+     * notifications for the main database (and the Database table      \
+     * in _Server if available).                                        \
+     *                                                                  \
+     * If we're monitoring the Database table and we get notified       \
+     * that the CS database has been deleted, we close the              \
+     * connection (which will restart the state machine). */            \
+    OVSDB_CS_STATE(MONITORING)                                          \
+                                                                        \
+    /* Terminal error state that indicates that nothing useful can be   \
+     * done, for example because the database server doesn't actually   \
+     * have the desired database.  We maintain the session with the     \
+     * database server anyway.  If it starts serving the database       \
+     * that we want, or if someone fixes and restarts the database,     \
+     * then it will kill the session and we will automatically          \
+     * reconnect and try again. */                                      \
+    OVSDB_CS_STATE(ERROR)                                               \
+                                                                        \
+    /* Terminal state that indicates we connected to a useless server   \
+     * in a cluster, e.g. one that is partitioned from the rest of      \
+     * the cluster. We're waiting to retry. */                          \
+    OVSDB_CS_STATE(RETRY)
+
+enum ovsdb_cs_state {
+#define OVSDB_CS_STATE(NAME) CS_S_##NAME,
+    OVSDB_CS_STATES
+#undef OVSDB_CS_STATE
+};
+
+static const char *
+ovsdb_cs_state_to_string(enum ovsdb_cs_state state)
+{
+    switch (state) {
+#define OVSDB_CS_STATE(NAME) case CS_S_##NAME: return #NAME;
+        OVSDB_CS_STATES
+#undef OVSDB_CS_STATE
+    default: return "<unknown>";
+    }
+}
+
+/* A database being monitored.
+ *
+ * There are two instances of this data structure for each CS instance, one for
+ * the _Server database used for working with clusters, and the other one for
+ * the actual database that the client is interested in.  */
+struct ovsdb_cs_db {
+    struct ovsdb_cs *cs;
+
+    /* Data. */
+    const char *db_name;        /* Database's name. */
+    struct hmap tables;         /* Contains "struct ovsdb_cs_db_table *"s.*/
+    struct json *monitor_id;
+    struct json *schema;
+
+    /* Monitor version. */
+    int max_version;            /* Maximum version of monitor request to use. */
+    int monitor_version;        /* 0 if not monitoring, 1=monitor,
+                                 * 2=monitor_cond, 3=monitor_cond_since. */
+
+    /* Condition changes. */
+    bool cond_changed;          /* Change not yet sent to server? */
+    unsigned int cond_seqno;    /* Increments when condition changes. */
+
+    /* Database locking. */
+    char *lock_name;            /* Name of lock we need, NULL if none. */
+    bool has_lock;              /* Has db server told us we have the lock? */
+    bool is_lock_contended;     /* Has db server told us we can't get lock? */
+    struct json *lock_request_id; /* JSON-RPC ID of in-flight lock request. */
+
+    /* Last db txn id, used for fast resync through monitor_cond_since */
+    struct uuid last_id;
+
+    /* Client interface. */
+    struct ovs_list events;
+    const struct ovsdb_cs_ops *ops;
+    void *ops_aux;
+};
+
+static const struct ovsdb_cs_ops ovsdb_cs_server_ops;
+
+static void ovsdb_cs_db_destroy_tables(struct ovsdb_cs_db *);
+static unsigned int ovsdb_cs_db_set_condition(
+    struct ovsdb_cs_db *, const char *db_name, const struct json *condition);
+
+static void ovsdb_cs_send_schema_request(struct ovsdb_cs *,
+                                          struct ovsdb_cs_db *);
+static void ovsdb_cs_send_db_change_aware(struct ovsdb_cs *);
+static bool ovsdb_cs_check_server_db(struct ovsdb_cs *);
+static void ovsdb_cs_clear_server_rows(struct ovsdb_cs *);
+static void ovsdb_cs_send_monitor_request(struct ovsdb_cs *,
+                                          struct ovsdb_cs_db *, int version);
+static void ovsdb_cs_db_ack_condition(struct ovsdb_cs_db *db);
+static void ovsdb_cs_db_sync_condition(struct ovsdb_cs_db *db);
+
+struct ovsdb_cs {
+    struct ovsdb_cs_db server;
+    struct ovsdb_cs_db data;
+
+    /* Session state.
+     *
+     * 'state_seqno' is a snapshot of the session's sequence number as returned
+     * jsonrpc_session_get_seqno(session), so if it differs from the value that
+     * function currently returns then the session has reconnected and the
+     * state machine must restart.  */
+    struct jsonrpc_session *session; /* Connection to the server. */
+    char *remote;                    /* 'session' remote name. */
+    enum ovsdb_cs_state state;       /* Current session state. */
+    unsigned int state_seqno;        /* See above. */
+    struct json *request_id;         /* JSON ID for request awaiting reply. */
+
+    /* IDs of outstanding transactions. */
+    struct json **txns;
+    size_t n_txns, allocated_txns;
+
+    /* Info for the _Server database. */
+    struct uuid cid;
+    struct hmap server_rows;
+
+    /* Clustered servers. */
+    uint64_t min_index;      /* Minimum allowed index, to avoid regression. */
+    bool leader_only;        /* If true, do not connect to Raft followers. */
+    bool shuffle_remotes;    /* If true, connect to servers in random order. */
+};
+
+static void ovsdb_cs_transition_at(struct ovsdb_cs *, enum ovsdb_cs_state,
+                                    const char *where);
+#define ovsdb_cs_transition(CS, STATE) \
+    ovsdb_cs_transition_at(CS, STATE, OVS_SOURCE_LOCATOR)
+
+static void ovsdb_cs_retry_at(struct ovsdb_cs *, const char *where);
+#define ovsdb_cs_retry(CS) ovsdb_cs_retry_at(CS, OVS_SOURCE_LOCATOR)
+
+static struct vlog_rate_limit syntax_rl = VLOG_RATE_LIMIT_INIT(1, 5);
+
+static void ovsdb_cs_db_parse_monitor_reply(struct ovsdb_cs_db *,
+                                            const struct json *result,
+                                            int version);
+static bool ovsdb_cs_db_parse_update_rpc(struct ovsdb_cs_db *,
+                                         const struct jsonrpc_msg *);
+static bool ovsdb_cs_handle_monitor_canceled(struct ovsdb_cs *,
+                                              struct ovsdb_cs_db *,
+                                              const struct jsonrpc_msg *);
+
+static bool ovsdb_cs_db_process_lock_replies(struct ovsdb_cs_db *,
+                                              const struct jsonrpc_msg *);
+static struct jsonrpc_msg *ovsdb_cs_db_compose_lock_request(
+    struct ovsdb_cs_db *);
+static struct jsonrpc_msg *ovsdb_cs_db_compose_unlock_request(
+    struct ovsdb_cs_db *);
+static void ovsdb_cs_db_parse_lock_reply(struct ovsdb_cs_db *,
+                                          const struct json *);
+static bool ovsdb_cs_db_parse_lock_notify(struct ovsdb_cs_db *,
+                                           const struct json *params,
+                                           bool new_has_lock);
+static void ovsdb_cs_send_cond_change(struct ovsdb_cs *);
+
+static bool ovsdb_cs_db_txn_process_reply(struct ovsdb_cs *,
+                                          const struct jsonrpc_msg *reply);
+
+/* Events. */
+
+void
+ovsdb_cs_event_destroy(struct ovsdb_cs_event *event)
+{
+    if (event) {
+        switch (event->type) {
+        case OVSDB_CS_EVENT_TYPE_RECONNECT:
+        case OVSDB_CS_EVENT_TYPE_LOCKED:
+            break;
+
+        case OVSDB_CS_EVENT_TYPE_UPDATE:
+            json_destroy(event->update.table_updates);
+            break;
+
+        case OVSDB_CS_EVENT_TYPE_TXN_REPLY:
+            jsonrpc_msg_destroy(event->txn_reply);
+            break;
+        }
+        free(event);
+    }
+}
+
+/* Lifecycle. */
+
+static void
+ovsdb_cs_db_init(struct ovsdb_cs_db *db, const char *db_name,
+                 struct ovsdb_cs *parent, int max_version,
+                 const struct ovsdb_cs_ops *ops, void *ops_aux)
+{
+    *db = (struct ovsdb_cs_db) {
+        .cs = parent,
+        .db_name = xstrdup(db_name),
+        .tables = HMAP_INITIALIZER(&db->tables),
+        .max_version = max_version,
+        .monitor_id = json_array_create_2(json_string_create("monid"),
+                                          json_string_create(db_name)),
+        .events = OVS_LIST_INITIALIZER(&db->events),
+        .ops = ops,
+        .ops_aux = ops_aux,
+    };
+}
+
+/* Creates and returns a new client synchronization object.  The connection
+ * will monitor remote database 'db_name'.  If 'retry' is true, then also
+ * reconnect if the connection fails.
+ *
+ * XXX 'max_version' should ordinarily be 3, to allow use of the most efficient
+ * "monitor_cond_since" method with the database.  Currently there's some kind
+ * of bug in the DDlog Rust code that interfaces to that, so instead
+ * ovn-northd-ddlog passes 1 to use plain 'monitor' instead.  Once the DDlog
+ * Rust code gets fixed, we might as well just delete 'max_version'
+ * entirely.
+ *
+ * 'ops' is a struct for northd_cs_run() to use, and 'ops_aux' is a pointer
+ * that gets passed into each call.
+ *
+ * Use ovsdb_cs_set_remote() to configure the database to which to connect.
+ * Until a remote is configured, no data can be retrieved.
+ */
+struct ovsdb_cs *
+ovsdb_cs_create(const char *db_name, int max_version,
+                const struct ovsdb_cs_ops *ops, void *ops_aux)
+{
+    struct ovsdb_cs *cs = xzalloc(sizeof *cs);
+    ovsdb_cs_db_init(&cs->server, "_Server", cs, 2, &ovsdb_cs_server_ops, cs);
+    ovsdb_cs_db_init(&cs->data, db_name, cs, max_version, ops, ops_aux);
+    cs->state_seqno = UINT_MAX;
+    cs->request_id = NULL;
+    cs->leader_only = true;
+    cs->shuffle_remotes = true;
+    hmap_init(&cs->server_rows);
+
+    return cs;
+}
+
+static void
+ovsdb_cs_db_destroy(struct ovsdb_cs_db *db)
+{
+    ovsdb_cs_db_destroy_tables(db);
+
+    json_destroy(db->monitor_id);
+    json_destroy(db->schema);
+
+    free(db->lock_name);
+
+    json_destroy(db->lock_request_id);
+
+    /* This list always gets flushed out at the end of ovsdb_cs_run(). */
+    ovs_assert(ovs_list_is_empty(&db->events));
+}
+
+/* Destroys 'cs' and all of the data structures that it manages. */
+void
+ovsdb_cs_destroy(struct ovsdb_cs *cs)
+{
+    if (cs) {
+        ovsdb_cs_db_destroy(&cs->server);
+        ovsdb_cs_db_destroy(&cs->data);
+        jsonrpc_session_close(cs->session);
+        free(cs->remote);
+        json_destroy(cs->request_id);
+
+        for (size_t i = 0; i < cs->n_txns; i++) {
+            json_destroy(cs->txns[i]);
+        }
+        free(cs->txns);
+
+        ovsdb_cs_clear_server_rows(cs);
+        hmap_destroy(&cs->server_rows);
+
+        free(cs);
+    }
+}
+
+static void
+ovsdb_cs_transition_at(struct ovsdb_cs *cs, enum ovsdb_cs_state new_state,
+                        const char *where)
+{
+    VLOG_DBG("%s: %s -> %s at %s",
+             cs->session ? jsonrpc_session_get_name(cs->session) : "void",
+             ovsdb_cs_state_to_string(cs->state),
+             ovsdb_cs_state_to_string(new_state),
+             where);
+    cs->state = new_state;
+}
+
+static void
+ovsdb_cs_send_request(struct ovsdb_cs *cs, struct jsonrpc_msg *request)
+{
+    json_destroy(cs->request_id);
+    cs->request_id = json_clone(request->id);
+    if (cs->session) {
+        jsonrpc_session_send(cs->session, request);
+    } else {
+        jsonrpc_msg_destroy(request);
+    }
+}
+
+static void
+ovsdb_cs_retry_at(struct ovsdb_cs *cs, const char *where)
+{
+    ovsdb_cs_force_reconnect(cs);
+    ovsdb_cs_transition_at(cs, CS_S_RETRY, where);
+}
+
+static void
+ovsdb_cs_restart_fsm(struct ovsdb_cs *cs)
+{
+    /* Resync data DB table conditions to avoid missing updates due to
+     * conditions that were in flight or changed locally while the connection
+     * was down.
+     */
+    ovsdb_cs_db_sync_condition(&cs->data);
+
+    ovsdb_cs_send_schema_request(cs, &cs->server);
+    ovsdb_cs_transition(cs, CS_S_SERVER_SCHEMA_REQUESTED);
+    cs->data.monitor_version = 0;
+    cs->server.monitor_version = 0;
+}
+
+static void
+ovsdb_cs_process_response(struct ovsdb_cs *cs, struct jsonrpc_msg *msg)
+{
+    bool ok = msg->type == JSONRPC_REPLY;
+    if (!ok
+        && cs->state != CS_S_SERVER_SCHEMA_REQUESTED
+        && cs->state != CS_S_SERVER_MONITOR_REQUESTED
+        && cs->state != CS_S_DATA_MONITOR_COND_REQUESTED
+        && cs->state != CS_S_DATA_MONITOR_COND_SINCE_REQUESTED) {
+        static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
+        char *s = jsonrpc_msg_to_string(msg);
+        VLOG_INFO_RL(&rl, "%s: received unexpected %s response in "
+                     "%s state: %s", jsonrpc_session_get_name(cs->session),
+                     jsonrpc_msg_type_to_string(msg->type),
+                     ovsdb_cs_state_to_string(cs->state),
+                     s);
+        free(s);
+        ovsdb_cs_retry(cs);
+        return;
+    }
+
+    switch (cs->state) {
+    case CS_S_SERVER_SCHEMA_REQUESTED:
+        if (ok) {
+            json_destroy(cs->server.schema);
+            cs->server.schema = json_clone(msg->result);
+            ovsdb_cs_send_monitor_request(cs, &cs->server,
+                                          cs->server.max_version);
+            ovsdb_cs_transition(cs, CS_S_SERVER_MONITOR_REQUESTED);
+        } else {
+            ovsdb_cs_send_schema_request(cs, &cs->data);
+            ovsdb_cs_transition(cs, CS_S_DATA_SCHEMA_REQUESTED);
+        }
+        break;
+
+    case CS_S_SERVER_MONITOR_REQUESTED:
+        if (ok) {
+            cs->server.monitor_version = cs->server.max_version;
+            ovsdb_cs_db_parse_monitor_reply(&cs->server, msg->result,
+                                            cs->server.monitor_version);
+            if (ovsdb_cs_check_server_db(cs)) {
+                ovsdb_cs_send_db_change_aware(cs);
+            }
+        } else {
+            ovsdb_cs_send_schema_request(cs, &cs->data);
+            ovsdb_cs_transition(cs, CS_S_DATA_SCHEMA_REQUESTED);
+        }
+        break;
+
+    case CS_S_DATA_SCHEMA_REQUESTED:
+        json_destroy(cs->data.schema);
+        cs->data.schema = json_clone(msg->result);
+        if (cs->data.max_version >= 2) {
+            ovsdb_cs_send_monitor_request(cs, &cs->data, 2);
+            ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_COND_REQUESTED);
+        } else {
+            ovsdb_cs_send_monitor_request(cs, &cs->data, 1);
+            ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_REQUESTED);
+        }
+        break;
+
+    case CS_S_DATA_MONITOR_COND_SINCE_REQUESTED:
+        if (!ok) {
+            /* "monitor_cond_since" not supported.  Try "monitor_cond". */
+            ovsdb_cs_send_monitor_request(cs, &cs->data, 2);
+            ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_COND_REQUESTED);
+        } else {
+            cs->data.monitor_version = 3;
+            ovsdb_cs_transition(cs, CS_S_MONITORING);
+            ovsdb_cs_db_parse_monitor_reply(&cs->data, msg->result, 3);
+        }
+        break;
+
+    case CS_S_DATA_MONITOR_COND_REQUESTED:
+        if (!ok) {
+            /* "monitor_cond" not supported.  Try "monitor". */
+            ovsdb_cs_send_monitor_request(cs, &cs->data, 1);
+            ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_REQUESTED);
+        } else {
+            cs->data.monitor_version = 2;
+            ovsdb_cs_transition(cs, CS_S_MONITORING);
+            ovsdb_cs_db_parse_monitor_reply(&cs->data, msg->result, 2);
+        }
+        break;
+
+    case CS_S_DATA_MONITOR_REQUESTED:
+        cs->data.monitor_version = 1;
+        ovsdb_cs_transition(cs, CS_S_MONITORING);
+        ovsdb_cs_db_parse_monitor_reply(&cs->data, msg->result, 1);
+        break;
+
+    case CS_S_MONITORING:
+        /* We don't normally have a request outstanding in this state.  If we
+         * do, it's a "monitor_cond_change", which means that the conditional
+         * monitor clauses were updated.
+         *
+         * Mark the last requested conditions as acked and if further
+         * condition changes were pending, send them now. */
+        ovsdb_cs_db_ack_condition(&cs->data);
+        ovsdb_cs_send_cond_change(cs);
+        cs->data.cond_seqno++;
+        break;
+
+    case CS_S_ERROR:
+    case CS_S_RETRY:
+        /* Nothing to do in this state. */
+        break;
+
+    default:
+        OVS_NOT_REACHED();
+    }
+}
+
+static void
+ovsdb_cs_process_msg(struct ovsdb_cs *cs, struct jsonrpc_msg *msg)
+{
+    bool is_response = (msg->type == JSONRPC_REPLY ||
+                        msg->type == JSONRPC_ERROR);
+
+    /* Process a reply to an outstanding request. */
+    if (is_response
+        && cs->request_id && json_equal(cs->request_id, msg->id)) {
+        json_destroy(cs->request_id);
+        cs->request_id = NULL;
+        ovsdb_cs_process_response(cs, msg);
+        return;
+    }
+
+    /* Process database contents updates. */
+    if (ovsdb_cs_db_parse_update_rpc(&cs->data, msg)) {
+        return;
+    }
+    if (cs->server.monitor_version
+        && ovsdb_cs_db_parse_update_rpc(&cs->server, msg)) {
+        ovsdb_cs_check_server_db(cs);
+        return;
+    }
+
+    if (ovsdb_cs_handle_monitor_canceled(cs, &cs->data, msg)
+        || (cs->server.monitor_version
+            && ovsdb_cs_handle_monitor_canceled(cs, &cs->server, msg))) {
+        return;
+    }
+
+    /* Process "lock" replies and related notifications. */
+    if (ovsdb_cs_db_process_lock_replies(&cs->data, msg)) {
+        return;
+    }
+
+    /* Process response to a database transaction we submitted. */
+    if (is_response && ovsdb_cs_db_txn_process_reply(cs, msg)) {
+        return;
+    }
+
+    /* Unknown message.  Log at a low level because this can happen if
+     * ovsdb_cs_txn_destroy() is called to destroy a transaction
+     * before we receive the reply.
+     *
+     * (We could sort those out from other kinds of unknown messages by
+     * using distinctive IDs for transactions, if it seems valuable to
+     * do so, and then it would be possible to use different log
+     * levels. XXX?) */
+    char *s = jsonrpc_msg_to_string(msg);
+    VLOG_DBG("%s: received unexpected %s message: %s",
+             jsonrpc_session_get_name(cs->session),
+             jsonrpc_msg_type_to_string(msg->type), s);
+    free(s);
+}
+
+static struct ovsdb_cs_event *
+ovsdb_cs_db_add_event(struct ovsdb_cs_db *db, enum ovsdb_cs_event_type type)
+{
+    struct ovsdb_cs_event *event = xmalloc(sizeof *event);
+    event->type = type;
+    ovs_list_push_back(&db->events, &event->list_node);
+    return event;
+}
+
+/* Processes a batch of messages from the database server on 'cs'.  This may
+ * cause the CS's contents to change.
+ *
+ * Initializes 'events' with a list of events that occurred on 'cs'.  The
+ * caller must process and destroy all of the events. */
+void
+ovsdb_cs_run(struct ovsdb_cs *cs, struct ovs_list *events)
+{
+    ovs_list_init(events);
+    if (!cs->session) {
+        return;
+    }
+
+    ovsdb_cs_send_cond_change(cs);
+
+    jsonrpc_session_run(cs->session);
+
+    unsigned int seqno = jsonrpc_session_get_seqno(cs->session);
+    if (cs->state_seqno != seqno) {
+        cs->state_seqno = seqno;
+        ovsdb_cs_restart_fsm(cs);
+
+        for (size_t i = 0; i < cs->n_txns; i++) {
+            json_destroy(cs->txns[i]);
+        }
+        cs->n_txns = 0;
+
+        ovsdb_cs_db_add_event(&cs->data, OVSDB_CS_EVENT_TYPE_RECONNECT);
+
+        if (cs->data.lock_name) {
+            jsonrpc_session_send(
+                cs->session,
+                ovsdb_cs_db_compose_lock_request(&cs->data));
+        }
+    }
+
+    for (int i = 0; i < 50; i++) {
+        struct jsonrpc_msg *msg = jsonrpc_session_recv(cs->session);
+        if (!msg) {
+            break;
+        }
+        ovsdb_cs_process_msg(cs, msg);
+        jsonrpc_msg_destroy(msg);
+    }
+    ovs_list_push_back_all(events, &cs->data.events);
+}
+
+/* Arranges for poll_block() to wake up when ovsdb_cs_run() has something to
+ * do or when activity occurs on a transaction on 'cs'. */
+void
+ovsdb_cs_wait(struct ovsdb_cs *cs)
+{
+    if (!cs->session) {
+        return;
+    }
+    jsonrpc_session_wait(cs->session);
+    jsonrpc_session_recv_wait(cs->session);
+}
+
+/* Network connection. */
+
+/* Changes the remote and creates a new session.
+ *
+ * If 'retry' is true, the connection to the remote will automatically retry
+ * when it fails.  If 'retry' is false, the connection is one-time. */
+void
+ovsdb_cs_set_remote(struct ovsdb_cs *cs, const char *remote, bool retry)
+{
+    if (cs
+        && ((remote != NULL) != (cs->remote != NULL)
+            || (remote && cs->remote && strcmp(remote, cs->remote)))) {
+        /* Close the old session, if any. */
+        if (cs->session) {
+            jsonrpc_session_close(cs->session);
+            cs->session = NULL;
+
+            free(cs->remote);
+            cs->remote = NULL;
+        }
+
+        /* Open new session, if any. */
+        if (remote) {
+            struct svec remotes = SVEC_EMPTY_INITIALIZER;
+            ovsdb_session_parse_remote(remote, &remotes, &cs->cid);
+            if (cs->shuffle_remotes) {
+                svec_shuffle(&remotes);
+            }
+            cs->session = jsonrpc_session_open_multiple(&remotes, retry);
+            svec_destroy(&remotes);
+
+            cs->state_seqno = UINT_MAX;
+
+            cs->remote = xstrdup(remote);
+        }
+    }
+}
+
+/* Reconfigures 'cs' so that it would reconnect to the database, if
+ * connection was dropped. */
+void
+ovsdb_cs_enable_reconnect(struct ovsdb_cs *cs)
+{
+    if (cs->session) {
+        jsonrpc_session_enable_reconnect(cs->session);
+    }
+}
+
+/* Forces 'cs' to drop its connection to the database and reconnect.  In the
+ * meantime, the contents of 'cs' will not change. */
+void
+ovsdb_cs_force_reconnect(struct ovsdb_cs *cs)
+{
+    if (cs->session) {
+        jsonrpc_session_force_reconnect(cs->session);
+    }
+}
+
+/* Drops 'cs''s current connection and the cached session.  This is useful if
+ * the client notices some kind of inconsistency. */
+void
+ovsdb_cs_flag_inconsistency(struct ovsdb_cs *cs)
+{
+    cs->data.last_id = UUID_ZERO;
+    ovsdb_cs_retry(cs);
+}
+
+/* Returns true if 'cs' is currently connected or will eventually try to
+ * reconnect. */
+bool
+ovsdb_cs_is_alive(const struct ovsdb_cs *cs)
+{
+    return (cs->session
+            && jsonrpc_session_is_alive(cs->session)
+            && cs->state != CS_S_ERROR);
+}
+
+/* Returns true if 'cs' is currently connected to a server. */
+bool
+ovsdb_cs_is_connected(const struct ovsdb_cs *cs)
+{
+    return cs->session && jsonrpc_session_is_connected(cs->session);
+}
+
+/* Returns the last error reported on a connection by 'cs'.  The return value
+ * is 0 only if no connection made by 'cs' has ever encountered an error and
+ * a negative response to a schema request has never been received. See
+ * jsonrpc_get_status() for jsonrpc_session_get_last_error() return value
+ * interpretation. */
+int
+ovsdb_cs_get_last_error(const struct ovsdb_cs *cs)
+{
+    int err = cs->session ? jsonrpc_session_get_last_error(cs->session) : 0;
+    if (err) {
+        return err;
+    } else if (cs->state == CS_S_ERROR) {
+        return ENOENT;
+    } else {
+        return 0;
+    }
+}
+
+/* Sets the "probe interval" for 'cs''s current session to 'probe_interval', in
+ * milliseconds. */
+void
+ovsdb_cs_set_probe_interval(const struct ovsdb_cs *cs, int probe_interval)
+{
+    if (cs->session) {
+        jsonrpc_session_set_probe_interval(cs->session, probe_interval);
+    }
+}
+
+/* Conditional monitoring. */
+
+/* A table being monitored.
+ *
+ * At the CS layer, the only thing we care about, table-wise, is the conditions
+ * we're using for monitoring them, so there's little here.  We only create
+ * these table structures at all for tables that have conditions. */
+struct ovsdb_cs_db_table {
+    struct hmap_node hmap_node; /* Indexed by 'name'. */
+    const char *name;      /* Table name. */
+
+    /* Each of these is a null pointer if it is empty, or JSON [<condition>*]
+     * or [true] or [false] otherwise.  [true] could be represented as a null
+     * pointer, but we want to distinguish "empty slot" from "a condition that
+     * is always true" in a slot. */
+    struct json *ack_cond; /* Last condition acked by the server. */
+    struct json *req_cond; /* Last condition requested to the server. */
+    struct json *new_cond; /* Latest condition set by the IDL client. */
+};
+
+/* A kind of condition, so that we can treat equivalent JSON as equivalent. */
+enum condition_type {
+    COND_FALSE,                 /* [] or [false] */
+    COND_TRUE,                  /* Null pointer or [true] */
+    COND_OTHER                  /* Anything else. */
+};
+
+/* Returns the condition_type for 'condition'. */
+static enum condition_type
+condition_classify(const struct json *condition)
+{
+    if (condition) {
+        const struct json_array *a = json_array(condition);
+        switch (a->n) {
+        case 0:
+            return COND_FALSE;
+
+        case 1:
+            return (a->elems[0]->type == JSON_FALSE ? COND_FALSE
+                    : a->elems[0]->type == JSON_TRUE ? COND_TRUE
+                    : COND_OTHER);
+
+        default:
+            return COND_OTHER;
+        }
+    } else {
+        return COND_TRUE;
+    }
+}
+
+/* Returns true if 'a' and 'b' are the same condition (in an obvious way; we're
+ * not going to compare for boolean equivalence or anything). */
+static bool
+condition_equal(const struct json *a, const struct json *b)
+{
+    enum condition_type type = condition_classify(a);
+    return (type == condition_classify(b)
+            && (type != COND_OTHER || json_equal(a, b)));
+}
+
+/* Returns a clone of 'condition', translating always-true and always-false to
+ * [true] and [false], respectively. */
+static struct json *
+condition_clone(const struct json *condition)
+{
+    switch (condition_classify(condition)) {
+    case COND_TRUE:
+        return json_array_create_1(json_boolean_create(true));
+
+    case COND_FALSE:
+        return json_array_create_1(json_boolean_create(false));
+
+    case COND_OTHER:
+        return json_clone(condition);
+    }
+
+    OVS_NOT_REACHED();
+}
+
+/* Returns the ovsdb_cs_db_table associated with 'table' in 'db', creating an
+ * empty one if necessary. */
+static struct ovsdb_cs_db_table *
+ovsdb_cs_db_get_table(struct ovsdb_cs_db *db, const char *table)
+{
+    uint32_t hash = hash_string(table, 0);
+    struct ovsdb_cs_db_table *t;
+
+    HMAP_FOR_EACH_WITH_HASH (t, hmap_node, hash, &db->tables) {
+        if (!strcmp(t->name, table)) {
+            return t;
+        }
+    }
+
+    t = xzalloc(sizeof *t);
+    t->name = xstrdup(table);
+    t->new_cond = json_array_create_1(json_boolean_create(true));
+    hmap_insert(&db->tables, &t->hmap_node, hash);
+    return t;
+}
+
+static void
+ovsdb_cs_db_destroy_tables(struct ovsdb_cs_db *db)
+{
+    struct ovsdb_cs_db_table *table, *next;
+    HMAP_FOR_EACH_SAFE (table, next, hmap_node, &db->tables) {
+        json_destroy(table->ack_cond);
+        json_destroy(table->req_cond);
+        json_destroy(table->new_cond);
+        hmap_remove(&db->tables, &table->hmap_node);
+    }
+    hmap_destroy(&db->tables);
+}
+
+static unsigned int
+ovsdb_cs_db_set_condition(struct ovsdb_cs_db *db, const char *table,
+                          const struct json *condition)
+{
+    /* Compare the new condition to the last known condition which can be
+     * either "new" (not sent yet), "requested" or "acked", in this order. */
+    struct ovsdb_cs_db_table *t = ovsdb_cs_db_get_table(db, table);
+    const struct json *table_cond = (t->new_cond ? t->new_cond
+                                     : t->req_cond ? t->req_cond
+                                     : t->ack_cond);
+    if (!condition_equal(condition, table_cond)) {
+        json_destroy(t->new_cond);
+        t->new_cond = condition_clone(condition);
+        db->cond_changed = true;
+        poll_immediate_wake();
+        return db->cond_seqno + 1;
+    } else if (table_cond != t->ack_cond) {
+        /* 'condition' was already set but has not been "acked" yet.  The IDL
+         * will be up to date when db->cond_seqno gets incremented. */
+        return db->cond_seqno + 1;
+    } else {
+        return db->cond_seqno;
+    }
+}
+
+/* Sets the replication condition for 'tc' in 'cs' to 'condition' and arranges
+ * to send the new condition to the database server.
+ *
+ * Return the next conditional update sequence number.  When this value and
+ * ovsdb_cs_get_condition_seqno() matches, 'cs' contains rows that match the
+ * 'condition'. */
+unsigned int
+ovsdb_cs_set_condition(struct ovsdb_cs *cs, const char *table,
+                       const struct json *condition)
+{
+    return ovsdb_cs_db_set_condition(&cs->data, table, condition);
+}
+
+/* Returns a "sequence number" that represents the number of conditional
+ * monitoring updates successfully received by the OVSDB server of a CS
+ * connection.
+ *
+ * ovsdb_cs_set_condition() sets a new condition that is different from the
+ * current condtion, the next expected "sequence number" is returned.
+ *
+ * Whenever ovsdb_cs_get_condition_seqno() returns a value that matches the
+ * return value of ovsdb_cs_set_condition(), the client is assured that:
+ *
+ *   - The ovsdb_cs_set_condition() changes has been acknowledged by the OVSDB
+ *     server.
+ *
+ *   -  'cs' now contains the content matches the new conditions.   */
+unsigned int
+ovsdb_cs_get_condition_seqno(const struct ovsdb_cs *cs)
+{
+    return cs->data.cond_seqno;
+}
+
+static struct json *
+ovsdb_cs_create_cond_change_req(const struct json *cond)
+{
+    struct json *monitor_cond_change_request = json_object_create();
+    json_object_put(monitor_cond_change_request, "where", json_clone(cond));
+    return monitor_cond_change_request;
+}
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_compose_cond_change(struct ovsdb_cs_db *db)
+{
+    if (!db->cond_changed) {
+        return NULL;
+    }
+
+    struct json *monitor_cond_change_requests = NULL;
+    struct ovsdb_cs_db_table *table;
+    HMAP_FOR_EACH (table, hmap_node, &db->tables) {
+        /* Always use the most recent conditions set by the CS client when
+         * requesting monitor_cond_change, i.e., table->new_cond.
+         */
+        if (table->new_cond) {
+            struct json *req =
+                ovsdb_cs_create_cond_change_req(table->new_cond);
+            if (req) {
+                if (!monitor_cond_change_requests) {
+                    monitor_cond_change_requests = json_object_create();
+                }
+                json_object_put(monitor_cond_change_requests,
+                                table->name,
+                                json_array_create_1(req));
+            }
+            /* Mark the new condition as requested by moving it to req_cond.
+             * If there's already requested condition that's a bug.
+             */
+            ovs_assert(table->req_cond == NULL);
+            table->req_cond = table->new_cond;
+            table->new_cond = NULL;
+        }
+    }
+
+    if (!monitor_cond_change_requests) {
+        return NULL;
+    }
+
+    db->cond_changed = false;
+    struct json *params = json_array_create_3(json_clone(db->monitor_id),
+                                              json_clone(db->monitor_id),
+                                              monitor_cond_change_requests);
+    return jsonrpc_create_request("monitor_cond_change", params, NULL);
+}
+
+/* Marks all requested table conditions in 'db' as acked by the server.
+ * It should be called when the server replies to monitor_cond_change
+ * requests.
+ */
+static void
+ovsdb_cs_db_ack_condition(struct ovsdb_cs_db *db)
+{
+    struct ovsdb_cs_db_table *table;
+    HMAP_FOR_EACH (table, hmap_node, &db->tables) {
+        if (table->req_cond) {
+            json_destroy(table->ack_cond);
+            table->ack_cond = table->req_cond;
+            table->req_cond = NULL;
+        }
+    }
+}
+
+/* Should be called when the CS fsm is restarted and resyncs table conditions
+ * based on the state the DB is in:
+ * - if a non-zero last_id is available for the DB then upon reconnect
+ *   the CS should first request acked conditions to avoid missing updates
+ *   about records that were added before the transaction with
+ *   txn-id == last_id. If there were requested condition changes in flight
+ *   (i.e., req_cond not NULL) and the CS client didn't set new conditions
+ *   (i.e., new_cond is NULL) then move req_cond to new_cond to trigger a
+ *   follow up monitor_cond_change request.
+ * - if there's no last_id available for the DB then it's safe to use the
+ *   latest conditions set by the CS client even if they weren't acked yet.
+ */
+static void
+ovsdb_cs_db_sync_condition(struct ovsdb_cs_db *db)
+{
+    bool ack_all = uuid_is_zero(&db->last_id);
+    if (ack_all) {
+        db->cond_changed = false;
+    }
+
+    struct ovsdb_cs_db_table *table;
+    HMAP_FOR_EACH (table, hmap_node, &db->tables) {
+        /* When monitor_cond_since requests will be issued, the
+         * table->ack_cond condition will be added to the "where" clause".
+         * Follow up monitor_cond_change requests will use table->new_cond.
+         */
+        if (ack_all) {
+            if (table->new_cond) {
+                json_destroy(table->req_cond);
+                table->req_cond = table->new_cond;
+                table->new_cond = NULL;
+            }
+
+            if (table->req_cond) {
+                json_destroy(table->ack_cond);
+                table->ack_cond = table->req_cond;
+                table->req_cond = NULL;
+            }
+        } else {
+            /* If there was no "unsent" condition but instead a
+             * monitor_cond_change request was in flight, move table->req_cond
+             * to table->new_cond and set db->cond_changed to trigger a new
+             * monitor_cond_change request.
+             *
+             * However, if a new condition has been set by the CS client,
+             * monitor_cond_change will be sent anyway and will use the most
+             * recent table->new_cond so there's no need to update it here.
+             */
+            if (table->req_cond && !table->new_cond) {
+                table->new_cond = table->req_cond;
+                table->req_cond = NULL;
+                db->cond_changed = true;
+            }
+        }
+    }
+}
+
+static void
+ovsdb_cs_send_cond_change(struct ovsdb_cs *cs)
+{
+    /* When 'cs->request_id' is not NULL, there is an outstanding
+     * conditional monitoring update request that we have not heard
+     * from the server yet. Don't generate another request in this case. */
+    if (!jsonrpc_session_is_connected(cs->session)
+        || cs->data.monitor_version == 1
+        || cs->request_id) {
+        return;
+    }
+
+    struct jsonrpc_msg *msg = ovsdb_cs_db_compose_cond_change(&cs->data);
+    if (msg) {
+        cs->request_id = json_clone(msg->id);
+        jsonrpc_session_send(cs->session, msg);
+    }
+}
+
+/* Clustered servers. */
+
+/* By default, or if 'leader_only' is true, when 'cs' connects to a clustered
+ * database, the CS layer will avoid servers other than the cluster
+ * leader. This ensures that any data that it reads and reports is up-to-date.
+ * If 'leader_only' is false, the CS layer will accept any server in the
+ * cluster, which means that for read-only transactions it can report and act
+ * on stale data (transactions that modify the database are always serialized
+ * even with false 'leader_only').  Refer to Understanding Cluster Consistency
+ * in ovsdb(7) for more information. */
+void
+ovsdb_cs_set_leader_only(struct ovsdb_cs *cs, bool leader_only)
+{
+    cs->leader_only = leader_only;
+    if (leader_only && cs->server.monitor_version) {
+        ovsdb_cs_check_server_db(cs);
+    }
+}
+
+/* Set whether the order of remotes should be shuffled, when there is more than
+ * one remote.  The setting doesn't take effect until the next time when
+ * ovsdb_cs_set_remote() is called. */
+void
+ovsdb_cs_set_shuffle_remotes(struct ovsdb_cs *cs, bool shuffle)
+{
+    cs->shuffle_remotes = shuffle;
+}
+
+/* Reset min_index to 0. This prevents a situation where the client
+ * thinks all databases have stale data, when they actually have all
+ * been destroyed and rebuilt from scratch.
+ */
+void
+ovsdb_cs_reset_min_index(struct ovsdb_cs *cs)
+{
+    cs->min_index = 0;
+}
+
+/* Database locks. */
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_set_lock(struct ovsdb_cs_db *db, const char *lock_name)
+{
+    if (db->lock_name
+        && (!lock_name || strcmp(lock_name, db->lock_name))) {
+        /* Release previous lock. */
+        struct jsonrpc_msg *msg = ovsdb_cs_db_compose_unlock_request(db);
+        free(db->lock_name);
+        db->lock_name = NULL;
+        db->is_lock_contended = false;
+        return msg;
+    }
+
+    if (lock_name && !db->lock_name) {
+        /* Acquire new lock. */
+        db->lock_name = xstrdup(lock_name);
+        return ovsdb_cs_db_compose_lock_request(db);
+    }
+
+    return NULL;
+}
+
+/* If 'lock_name' is nonnull, configures 'cs' to obtain the named lock from the
+ * database server and to prevent modifying the database when the lock cannot
+ * be acquired (that is, when another client has the same lock).
+ *
+ * If 'lock_name' is NULL, drops the locking requirement and releases the
+ * lock. */
+void
+ovsdb_cs_set_lock(struct ovsdb_cs *cs, const char *lock_name)
+{
+    for (;;) {
+        struct jsonrpc_msg *msg = ovsdb_cs_db_set_lock(&cs->data, lock_name);
+        if (!msg) {
+            break;
+        }
+        if (cs->session) {
+            jsonrpc_session_send(cs->session, msg);
+        } else {
+            jsonrpc_msg_destroy(msg);
+        }
+    }
+}
+
+/* Returns the name of the lock that 'cs' is trying to obtain, or NULL if none
+ * is configured. */
+const char *
+ovsdb_cs_get_lock(const struct ovsdb_cs *cs)
+{
+    return cs->data.lock_name;
+}
+
+/* Returns true if 'cs' is configured to obtain a lock and owns that lock,
+ * false if it doesn't own the lock or isn't configured to obtain one.
+ *
+ * Locking and unlocking happens asynchronously from the database client's
+ * point of view, so the information is only useful for optimization (e.g. if
+ * the client doesn't have the lock then there's no point in trying to write to
+ * the database). */
+bool
+ovsdb_cs_has_lock(const struct ovsdb_cs *cs)
+{
+    return cs->data.has_lock;
+}
+
+/* Returns true if 'cs' is configured to obtain a lock but the database server
+ * has indicated that some other client already owns the requested lock. */
+bool
+ovsdb_cs_is_lock_contended(const struct ovsdb_cs *cs)
+{
+    return cs->data.is_lock_contended;
+}
+
+static void
+ovsdb_cs_db_update_has_lock(struct ovsdb_cs_db *db, bool new_has_lock)
+{
+    if (new_has_lock && !db->has_lock) {
+        ovsdb_cs_db_add_event(db, OVSDB_CS_EVENT_TYPE_LOCKED);
+        db->is_lock_contended = false;
+    }
+    db->has_lock = new_has_lock;
+}
+
+static bool
+ovsdb_cs_db_process_lock_replies(struct ovsdb_cs_db *db,
+                                  const struct jsonrpc_msg *msg)
+{
+    if (msg->type == JSONRPC_REPLY
+        && db->lock_request_id
+        && json_equal(db->lock_request_id, msg->id)) {
+        /* Reply to our "lock" request. */
+        ovsdb_cs_db_parse_lock_reply(db, msg->result);
+        return true;
+    }
+
+    if (msg->type == JSONRPC_NOTIFY) {
+        if (!strcmp(msg->method, "locked")) {
+            /* We got our lock. */
+            return ovsdb_cs_db_parse_lock_notify(db, msg->params, true);
+        } else if (!strcmp(msg->method, "stolen")) {
+            /* Someone else stole our lock. */
+            return ovsdb_cs_db_parse_lock_notify(db, msg->params, false);
+        }
+    }
+
+    return false;
+}
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_compose_lock_request__(struct ovsdb_cs_db *db,
+                                    const char *method)
+{
+    ovsdb_cs_db_update_has_lock(db, false);
+
+    json_destroy(db->lock_request_id);
+    db->lock_request_id = NULL;
+
+    struct json *params = json_array_create_1(json_string_create(
+                                                  db->lock_name));
+    return jsonrpc_create_request(method, params, NULL);
+}
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_compose_lock_request(struct ovsdb_cs_db *db)
+{
+    struct jsonrpc_msg *msg = ovsdb_cs_db_compose_lock_request__(db, "lock");
+    db->lock_request_id = json_clone(msg->id);
+    return msg;
+}
+
+static struct jsonrpc_msg *
+ovsdb_cs_db_compose_unlock_request(struct ovsdb_cs_db *db)
+{
+    return ovsdb_cs_db_compose_lock_request__(db, "unlock");
+}
+
+static void
+ovsdb_cs_db_parse_lock_reply(struct ovsdb_cs_db *db,
+                              const struct json *result)
+{
+    bool got_lock;
+
+    json_destroy(db->lock_request_id);
+    db->lock_request_id = NULL;
+
+    if (result->type == JSON_OBJECT) {
+        const struct json *locked;
+
+        locked = shash_find_data(json_object(result), "locked");
+        got_lock = locked && locked->type == JSON_TRUE;
+    } else {
+        got_lock = false;
+    }
+
+    ovsdb_cs_db_update_has_lock(db, got_lock);
+    if (!got_lock) {
+        db->is_lock_contended = true;
+    }
+}
+
+static bool
+ovsdb_cs_db_parse_lock_notify(struct ovsdb_cs_db *db,
+                               const struct json *params,
+                               bool new_has_lock)
+{
+    if (db->lock_name
+        && params->type == JSON_ARRAY
+        && json_array(params)->n > 0
+        && json_array(params)->elems[0]->type == JSON_STRING) {
+        const char *lock_name = json_string(json_array(params)->elems[0]);
+
+        if (!strcmp(db->lock_name, lock_name)) {
+            ovsdb_cs_db_update_has_lock(db, new_has_lock);
+            if (!new_has_lock) {
+                db->is_lock_contended = true;
+            }
+            return true;
+        }
+    }
+    return false;
+}
+
+/* Transactions. */
+
+static bool
+ovsdb_cs_db_txn_process_reply(struct ovsdb_cs *cs,
+                              const struct jsonrpc_msg *reply)
+{
+    bool found = ovsdb_cs_forget_transaction(cs, reply->id);
+    if (found) {
+        struct ovsdb_cs_event *event
+            = ovsdb_cs_db_add_event(&cs->data, OVSDB_CS_EVENT_TYPE_TXN_REPLY);
+        event->txn_reply = jsonrpc_msg_clone(reply);
+    }
+    return found;
+}
+
+/* Returns true if 'cs' can be sent a transaction now, false otherwise.  This
+ * is useful for optimization: there is no point in composing and sending a
+ * transaction if it returns false. */
+bool
+ovsdb_cs_may_send_transaction(const struct ovsdb_cs *cs)
+{
+    return (cs->session != NULL
+            && cs->state == CS_S_MONITORING
+            && (!cs->data.lock_name || ovsdb_cs_has_lock(cs)));
+}
+
+/* Attempts to send a transaction with the specified 'operations' to 'cs''s
+ * server.  On success, returns the request ID; the caller must eventually free
+ * it.  On failure, returns NULL. */
+struct json * OVS_WARN_UNUSED_RESULT
+ovsdb_cs_send_transaction(struct ovsdb_cs *cs, struct json *operations)
+{
+    if (!ovsdb_cs_may_send_transaction(cs)) {
+        json_destroy(operations);
+        return NULL;
+    }
+
+    if (cs->data.lock_name) {
+        struct json *assertion = json_object_create();
+        json_object_put_string(assertion, "op", "assert");
+        json_object_put_string(assertion, "lock", cs->data.lock_name);
+        json_array_add(operations, assertion);
+    }
+
+    struct json *request_id;
+    struct jsonrpc_msg *request = jsonrpc_create_request(
+        "transact", operations, &request_id);
+    int error = jsonrpc_session_send(cs->session, request);
+    if (error) {
+        json_destroy(request_id);
+        return NULL;
+    }
+
+    if (cs->n_txns >= cs->allocated_txns) {
+        cs->txns = x2nrealloc(cs->txns, &cs->allocated_txns,
+                              sizeof *cs->txns);
+    }
+    cs->txns[cs->n_txns++] = json_clone(request_id);
+    return request_id;
+}
+
+/* Makes 'cs' drop its record of transaction 'request_id'.  If a reply arrives
+ * for it later (which it will, unless the connection drops in the meantime),
+ * it won't be reported through an event.
+ *
+ * Returns true if 'request_id' was known, false otherwise. */
+bool
+ovsdb_cs_forget_transaction(struct ovsdb_cs *cs, const struct json *request_id)
+{
+    for (size_t i = 0; i < cs->n_txns; i++) {
+        if (json_equal(request_id, cs->txns[i])) {
+            cs->txns[i] = cs->txns[--cs->n_txns];
+            return true;
+        }
+    }
+    return false;
+}
+
+static void
+ovsdb_cs_send_schema_request(struct ovsdb_cs *cs,
+                              struct ovsdb_cs_db *db)
+{
+    ovsdb_cs_send_request(cs, jsonrpc_create_request(
+                               "get_schema",
+                               json_array_create_1(json_string_create(
+                                                       db->db_name)),
+                               NULL));
+}
+
+static void
+ovsdb_cs_send_db_change_aware(struct ovsdb_cs *cs)
+{
+    struct jsonrpc_msg *msg = jsonrpc_create_request(
+        "set_db_change_aware", json_array_create_1(json_boolean_create(true)),
+        NULL);
+    jsonrpc_session_send(cs->session, msg);
+}
+
+static void
+ovsdb_cs_send_monitor_request(struct ovsdb_cs *cs, struct ovsdb_cs_db *db,
+                              int version)
+{
+    struct json *mrs = db->ops->compose_monitor_requests(
+        db->schema, db->ops_aux);
+    /* XXX handle failure */
+    ovs_assert(mrs->type == JSON_OBJECT);
+
+    if (version > 1) {
+        struct ovsdb_cs_db_table *table;
+        HMAP_FOR_EACH (table, hmap_node, &db->tables) {
+            if (table->ack_cond) {
+                struct json *mr = shash_find_data(json_object(mrs),
+                                                  table->name);
+                if (!mr) {
+                    mr = json_array_create_empty();
+                    json_object_put(mrs, table->name, mr);
+                }
+                ovs_assert(mr->type == JSON_ARRAY);
+
+                struct json *mr0;
+                if (json_array(mr)->n == 0) {
+                    mr0 = json_object_create();
+                    json_object_put(mr0, "columns", json_array_create_empty());
+                    json_array_add(mr, mr0);
+                } else {
+                    mr0 = json_array(mr)->elems[0];
+                }
+                ovs_assert(mr0->type == JSON_OBJECT);
+
+                json_object_put(mr0, "where",
+                                json_clone(table->ack_cond));
+            }
+        }
+    }
+
+    const char *method = (version == 1 ? "monitor"
+                          : version == 2 ? "monitor_cond"
+                          : "monitor_cond_since");
+    struct json *params = json_array_create_3(
+                              json_string_create(db->db_name),
+                              json_clone(db->monitor_id),
+                              mrs);
+    if (version == 3) {
+        struct json *json_last_id = json_string_create_nocopy(
+            xasprintf(UUID_FMT, UUID_ARGS(&db->last_id)));
+        json_array_add(params, json_last_id);
+    }
+    ovsdb_cs_send_request(cs, jsonrpc_create_request(method, params, NULL));
+}
+
+static void
+log_parse_update_error(struct ovsdb_error *error)
+{
+    if (!VLOG_DROP_WARN(&syntax_rl)) {
+        char *s = ovsdb_error_to_string(error);
+        VLOG_WARN_RL(&syntax_rl, "%s", s);
+        free(s);
+    }
+    ovsdb_error_destroy(error);
+}
+
+static void
+ovsdb_cs_db_add_update(struct ovsdb_cs_db *db,
+                       const struct json *table_updates, int version,
+                       bool clear)
+{
+    struct ovsdb_cs_event *event = ovsdb_cs_db_add_event(
+        db, OVSDB_CS_EVENT_TYPE_UPDATE);
+    event->update = (struct ovsdb_cs_update_event) {
+        .table_updates = json_clone(table_updates),
+        .clear = clear,
+        .version = version,
+    };
+}
+
+static void
+ovsdb_cs_db_parse_monitor_reply(struct ovsdb_cs_db *db,
+                                const struct json *result, int version)
+{
+    const struct json *table_updates;
+    bool clear;
+    if (version == 3) {
+        struct uuid last_id;
+        if (result->type != JSON_ARRAY || result->array.n != 3
+            || (result->array.elems[0]->type != JSON_TRUE &&
+                result->array.elems[0]->type != JSON_FALSE)
+            || result->array.elems[1]->type != JSON_STRING
+            || !uuid_from_string(&last_id,
+                                 json_string(result->array.elems[1]))) {
+            struct ovsdb_error *error = ovsdb_syntax_error(
+                result, NULL, "bad monitor_cond_since reply format");
+            log_parse_update_error(error);
+            return;
+        }
+
+        bool found = json_boolean(result->array.elems[0]);
+        clear = !found;
+        table_updates = result->array.elems[2];
+    } else {
+        clear = true;
+        table_updates = result;
+    }
+
+    ovsdb_cs_db_add_update(db, table_updates, version, clear);
+}
+
+static bool
+ovsdb_cs_db_parse_update_rpc(struct ovsdb_cs_db *db,
+                             const struct jsonrpc_msg *msg)
+{
+    if (msg->type != JSONRPC_NOTIFY) {
+        return false;
+    }
+
+    int version = (!strcmp(msg->method, "update") ? 1
+                   : !strcmp(msg->method, "update2") ? 2
+                   : !strcmp(msg->method, "update3") ? 3
+                   : 0);
+    if (!version) {
+        return false;
+    }
+
+    struct json *params = msg->params;
+    int n = version == 3 ? 3 : 2;
+    if (params->type != JSON_ARRAY || params->array.n != n) {
+        struct ovsdb_error *error = ovsdb_syntax_error(
+            params, NULL, "%s must be an array with %u elements.",
+            msg->method, n);
+        log_parse_update_error(error);
+        return false;
+    }
+
+    if (!json_equal(params->array.elems[0], db->monitor_id)) {
+        return false;
+    }
+
+    if (version == 3) {
+        const char *last_id = json_string(params->array.elems[1]);
+        if (!uuid_from_string(&db->last_id, last_id)) {
+            struct ovsdb_error *error = ovsdb_syntax_error(
+                params, NULL, "Last-id %s is not in UUID format.", last_id);
+            log_parse_update_error(error);
+            return false;
+        }
+    }
+
+    struct json *table_updates = params->array.elems[version == 3 ? 2 : 1];
+    ovsdb_cs_db_add_update(db, table_updates, version, false);
+    return true;
+}
+
+static bool
+ovsdb_cs_handle_monitor_canceled(struct ovsdb_cs *cs,
+                                 struct ovsdb_cs_db *db,
+                                 const struct jsonrpc_msg *msg)
+{
+    if (msg->type != JSONRPC_NOTIFY
+        || strcmp(msg->method, "monitor_canceled")
+        || msg->params->type != JSON_ARRAY
+        || msg->params->array.n != 1
+        || !json_equal(msg->params->array.elems[0], db->monitor_id)) {
+        return false;
+    }
+
+    db->monitor_version = 0;
+
+    /* Cancel the other monitor and restart the FSM from the top.
+     *
+     * Maybe a more sophisticated response would be better in some cases, but
+     * it doesn't seem worth optimizing yet.  (Although this is already more
+     * sophisticated than just dropping the connection and reconnecting.) */
+    struct ovsdb_cs_db *other_db
+        = db == &cs->data ? &cs->server : &cs->data;
+    if (other_db->monitor_version) {
+        jsonrpc_session_send(
+            cs->session,
+            jsonrpc_create_request(
+                "monitor_cancel",
+                json_array_create_1(json_clone(other_db->monitor_id)), NULL));
+        other_db->monitor_version = 0;
+    }
+    ovsdb_cs_restart_fsm(cs);
+
+    return true;
+}
+
+/* The _Server database.
+ *
+ * We replicate the Database table in the _Server database because this is the
+ * only way to find out properties we need to know for clustering, such as
+ * whether a database is clustered at all and whether this server is the
+ * leader.
+ *
+ * This code implements a kind of simple IDL-like layer. */
+
+struct server_column {
+    const char *name;
+    struct ovsdb_type type;
+};
+enum server_column_index {
+    COL_NAME,
+    COL_MODEL,
+    COL_CONNECTED,
+    COL_LEADER,
+    COL_SCHEMA,
+    COL_CID,
+    COL_INDEX,
+};
+#define OPTIONAL_COLUMN(TYPE) \
+    {                                           \
+        .key = OVSDB_BASE_##TYPE##_INIT,        \
+        .value = OVSDB_BASE_VOID_INIT,          \
+        .n_min = 0,                             \
+        .n_max = 1                              \
+    }
+static const struct server_column server_columns[] = {
+    [COL_NAME] = {"name",  OPTIONAL_COLUMN(STRING) },
+    [COL_MODEL] = {"model", OPTIONAL_COLUMN(STRING) },
+    [COL_CONNECTED] = {"connected", OPTIONAL_COLUMN(BOOLEAN) },
+    [COL_LEADER] = {"leader", OPTIONAL_COLUMN(BOOLEAN) },
+    [COL_SCHEMA] = {"schema", OPTIONAL_COLUMN(STRING) },
+    [COL_CID] = {"cid", OPTIONAL_COLUMN(UUID) },
+    [COL_INDEX] = {"index", OPTIONAL_COLUMN(INTEGER) },
+};
+#define N_SERVER_COLUMNS ARRAY_SIZE(server_columns)
+struct server_row {
+    struct hmap_node hmap_node;
+    struct uuid uuid;
+    struct ovsdb_datum data[N_SERVER_COLUMNS];
+};
+
+static void
+server_row_destroy(struct server_row *row)
+{
+    if (row) {
+        for (size_t i = 0; i < N_SERVER_COLUMNS; i++) {
+            ovsdb_datum_destroy(&row->data[i], &server_columns[i].type);
+        }
+        free(row);
+    }
+}
+
+static struct server_row *
+ovsdb_cs_find_server_row(struct ovsdb_cs *cs, const struct uuid *uuid)
+{
+    struct server_row *row;
+    HMAP_FOR_EACH (row, hmap_node, &cs->server_rows) {
+        if (uuid_equals(uuid, &row->uuid)) {
+            return row;
+        }
+    }
+    return NULL;
+}
+
+static void
+ovsdb_cs_delete_server_row(struct ovsdb_cs *cs, struct server_row *row)
+{
+    hmap_remove(&cs->server_rows, &row->hmap_node);
+    server_row_destroy(row);
+}
+
+static struct server_row *
+ovsdb_cs_insert_server_row(struct ovsdb_cs *cs, const struct uuid *uuid)
+{
+    struct server_row *row = xmalloc(sizeof *row);
+    hmap_insert(&cs->server_rows, &row->hmap_node, uuid_hash(uuid));
+    row->uuid = *uuid;
+    for (size_t i = 0; i < N_SERVER_COLUMNS; i++) {
+        ovsdb_datum_init_default(&row->data[i], &server_columns[i].type);
+    }
+    return row;
+}
+
+static void
+ovsdb_cs_update_server_row(struct server_row *row,
+                           const struct shash *update, bool xor)
+{
+    for (size_t i = 0; i < N_SERVER_COLUMNS; i++) {
+        const struct server_column *column = &server_columns[i];
+        struct shash_node *node = shash_find(update, column->name);
+        if (!node) {
+            continue;
+        }
+        const struct json *json = node->data;
+
+        struct ovsdb_datum *old = &row->data[i];
+        struct ovsdb_datum new;
+        if (!xor) {
+            struct ovsdb_error *error = ovsdb_datum_from_json(
+                &new, &column->type, json, NULL);
+            if (error) {
+                ovsdb_error_destroy(error);
+                continue;
+            }
+        } else {
+            struct ovsdb_datum diff;
+            struct ovsdb_error *error = ovsdb_transient_datum_from_json(
+                &diff, &column->type, json);
+            if (error) {
+                ovsdb_error_destroy(error);
+                continue;
+            }
+
+            error = ovsdb_datum_apply_diff(&new, old, &diff, &column->type);
+            if (error) {
+                ovsdb_error_destroy(error);
+                ovsdb_datum_destroy(&new, &column->type);
+                continue;
+            }
+            ovsdb_datum_destroy(&diff, &column->type);
+        }
+
+        ovsdb_datum_destroy(&row->data[i], &column->type);
+        row->data[i] = new;
+    }
+}
+
+static void
+ovsdb_cs_clear_server_rows(struct ovsdb_cs *cs)
+{
+    struct server_row *row, *next;
+    HMAP_FOR_EACH_SAFE (row, next, hmap_node, &cs->server_rows) {
+        ovsdb_cs_delete_server_row(cs, row);
+    }
+}
+
+static void log_parse_update_error(struct ovsdb_error *);
+
+static void
+ovsdb_cs_process_server_event(struct ovsdb_cs *cs,
+                              const struct ovsdb_cs_event *event)
+{
+    ovs_assert(event->type == OVSDB_CS_EVENT_TYPE_UPDATE);
+
+    const struct ovsdb_cs_update_event *update = &event->update;
+    struct ovsdb_cs_db_update *du;
+    struct ovsdb_error *error = ovsdb_cs_parse_db_update(
+        update->table_updates, update->version, &du);
+    if (error) {
+        log_parse_update_error(error);
+        return;
+    }
+
+    if (update->clear) {
+        ovsdb_cs_clear_server_rows(cs);
+    }
+
+    const struct ovsdb_cs_table_update *tu = ovsdb_cs_db_update_find_table(
+        du, "Database");
+    if (tu) {
+        for (size_t i = 0; i < tu->n; i++) {
+            const struct ovsdb_cs_row_update *ru = &tu->row_updates[i];
+            struct server_row *row
+                = ovsdb_cs_find_server_row(cs, &ru->row_uuid);
+            if (ru->type == OVSDB_CS_ROW_DELETE) {
+                ovsdb_cs_delete_server_row(cs, row);
+            } else {
+                if (!row) {
+                    row = ovsdb_cs_insert_server_row(cs, &ru->row_uuid);
+                }
+                ovsdb_cs_update_server_row(row, ru->columns,
+                                           ru->type == OVSDB_CS_ROW_XOR);
+            }
+        }
+    }
+
+    ovsdb_cs_db_update_destroy(du);
+}
+
+static const char *
+server_column_get_string(const struct server_row *row,
+                         enum server_column_index index,
+                         const char *default_value)
+{
+    ovs_assert(server_columns[index].type.key.type == OVSDB_TYPE_STRING);
+    const struct ovsdb_datum *d = &row->data[index];
+    return d->n == 1 ? d->keys[0].string : default_value;
+}
+
+static bool
+server_column_get_bool(const struct server_row *row,
+                       enum server_column_index index,
+                       bool default_value)
+{
+    ovs_assert(server_columns[index].type.key.type == OVSDB_TYPE_BOOLEAN);
+    const struct ovsdb_datum *d = &row->data[index];
+    return d->n == 1 ? d->keys[0].boolean : default_value;
+}
+
+static uint64_t
+server_column_get_int(const struct server_row *row,
+                      enum server_column_index index,
+                      uint64_t default_value)
+{
+    ovs_assert(server_columns[index].type.key.type == OVSDB_TYPE_INTEGER);
+    const struct ovsdb_datum *d = &row->data[index];
+    return d->n == 1 ? d->keys[0].integer : default_value;
+}
+
+static const struct uuid *
+server_column_get_uuid(const struct server_row *row,
+                       enum server_column_index index,
+                       const struct uuid *default_value)
+{
+    ovs_assert(server_columns[index].type.key.type == OVSDB_TYPE_UUID);
+    const struct ovsdb_datum *d = &row->data[index];
+    return d->n == 1 ? &d->keys[0].uuid : default_value;
+}
+
+static const struct server_row *
+ovsdb_find_server_row(struct ovsdb_cs *cs)
+{
+    const struct server_row *row;
+    HMAP_FOR_EACH (row, hmap_node, &cs->server_rows) {
+        const struct uuid *cid = server_column_get_uuid(row, COL_CID, NULL);
+        const char *name = server_column_get_string(row, COL_NAME, NULL);
+        if (uuid_is_zero(&cs->cid)
+            ? (name && !strcmp(cs->data.db_name, name))
+            : (cid && uuid_equals(cid, &cs->cid))) {
+            return row;
+        }
+    }
+    return NULL;
+}
+
+static void OVS_UNUSED
+ovsdb_log_server_rows(const struct ovsdb_cs *cs)
+{
+    int row_num = 0;
+    const struct server_row *row;
+    HMAP_FOR_EACH (row, hmap_node, &cs->server_rows) {
+        struct ds s = DS_EMPTY_INITIALIZER;
+        for (size_t i = 0; i < N_SERVER_COLUMNS; i++) {
+            ds_put_format(&s, " %s=", server_columns[i].name);
+            if (i == COL_SCHEMA) {
+                ds_put_format(&s, "...");
+            } else {
+                ovsdb_datum_to_string(&row->data[i], &server_columns[i].type,
+                                      &s);
+            }
+        }
+        VLOG_INFO("row %d:%s", row_num++, ds_cstr(&s));
+        ds_destroy(&s);
+    }
+}
+
+static bool
+ovsdb_cs_check_server_db__(struct ovsdb_cs *cs)
+{
+    struct ovsdb_cs_event *event;
+    LIST_FOR_EACH_POP (event, list_node, &cs->server.events) {
+        ovsdb_cs_process_server_event(cs, event);
+        ovsdb_cs_event_destroy(event);
+    }
+
+    const struct server_row *db_row = ovsdb_find_server_row(cs);
+    static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
+    const char *server_name = jsonrpc_session_get_name(cs->session);
+    if (!db_row) {
+        VLOG_INFO_RL(&rl, "%s: server does not have %s database",
+                     server_name, cs->data.db_name);
+        return false;
+    }
+
+    bool ok = false;
+    const char *model = server_column_get_string(db_row, COL_MODEL, "");
+    const char *schema = server_column_get_string(db_row, COL_SCHEMA, NULL);
+    if (!strcmp(model, "clustered")) {
+        bool connected = server_column_get_bool(db_row, COL_CONNECTED, false);
+        bool leader = server_column_get_bool(db_row, COL_LEADER, false);
+        uint64_t index = server_column_get_int(db_row, COL_INDEX, 0);
+
+        if (!schema) {
+            VLOG_INFO("%s: clustered database server has not yet joined "
+                      "cluster; trying another server", server_name);
+        } else if (!connected) {
+            VLOG_INFO("%s: clustered database server is disconnected "
+                      "from cluster; trying another server", server_name);
+        } else if (cs->leader_only && !leader) {
+            VLOG_INFO("%s: clustered database server is not cluster "
+                      "leader; trying another server", server_name);
+        } else if (index < cs->min_index) {
+            VLOG_WARN("%s: clustered database server has stale data; "
+                      "trying another server", server_name);
+        } else {
+            cs->min_index = index;
+            ok = true;
+        }
+    } else {
+        if (!schema) {
+            VLOG_INFO("%s: missing database schema", server_name);
+        } else {
+            ok = true;
+        }
+    }
+    if (!ok) {
+        return false;
+    }
+
+    if (cs->state == CS_S_SERVER_MONITOR_REQUESTED) {
+        json_destroy(cs->data.schema);
+        cs->data.schema = json_from_string(schema);
+        if (cs->data.max_version >= 3) {
+            ovsdb_cs_send_monitor_request(cs, &cs->data, 3);
+            ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_COND_SINCE_REQUESTED);
+        } else if (cs->data.max_version >= 2) {
+            ovsdb_cs_send_monitor_request(cs, &cs->data, 2);
+            ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_COND_REQUESTED);
+        } else {
+            ovsdb_cs_send_monitor_request(cs, &cs->data, 1);
+            ovsdb_cs_transition(cs, CS_S_DATA_MONITOR_REQUESTED);
+        }
+    }
+    return true;
+}
+
+static bool
+ovsdb_cs_check_server_db(struct ovsdb_cs *cs)
+{
+    bool ok = ovsdb_cs_check_server_db__(cs);
+    if (!ok) {
+        ovsdb_cs_retry(cs);
+    }
+    return ok;
+}
+
+static struct json *
+ovsdb_cs_compose_server_monitor_request(const struct json *schema_json,
+                                        void *cs_)
+{
+    struct ovsdb_cs *cs = cs_;
+    struct shash *schema = ovsdb_cs_parse_schema(schema_json);
+    struct json *monitor_requests = json_object_create();
+
+    const char *table_name = "Database";
+    const struct sset *table_schema
+        = schema ? shash_find_data(schema, table_name) : NULL;
+    if (!table_schema) {
+        VLOG_WARN("%s database lacks %s table "
+                  "(database needs upgrade?)",
+                  cs->server.db_name, table_name);
+        /* XXX return failure? */
+    } else {
+        struct json *columns = json_array_create_empty();
+        for (size_t j = 0; j < N_SERVER_COLUMNS; j++) {
+            const struct server_column *column = &server_columns[j];
+            bool db_has_column = (table_schema &&
+                                  sset_contains(table_schema, column->name));
+            if (table_schema && !db_has_column) {
+                VLOG_WARN("%s table in %s database lacks %s column "
+                          "(database needs upgrade?)",
+                          table_name, cs->server.db_name, column->name);
+                continue;
+            }
+            json_array_add(columns, json_string_create(column->name));
+        }
+
+        struct json *monitor_request = json_object_create();
+        json_object_put(monitor_request, "columns", columns);
+        json_object_put(monitor_requests, table_name,
+                        json_array_create_1(monitor_request));
+    }
+    ovsdb_cs_free_schema(schema);
+
+    return monitor_requests;
+}
+
+static const struct ovsdb_cs_ops ovsdb_cs_server_ops = {
+    ovsdb_cs_compose_server_monitor_request
+};
 
 static void
 log_error(struct ovsdb_error *error)
@@ -324,3 +2263,17 @@ ovsdb_cs_db_update_destroy(struct ovsdb_cs_db_update *du)
     free(du->table_updates);
     free(du);
 }
+
+const struct ovsdb_cs_table_update *
+ovsdb_cs_db_update_find_table(const struct ovsdb_cs_db_update *du,
+                              const char *table_name)
+{
+    for (size_t i = 0; i < du->n; i++) {
+        const struct ovsdb_cs_table_update *tu = &du->table_updates[i];
+        if (!strcmp(tu->table_name, table_name)) {
+            return tu;
+        }
+    }
+    return NULL;
+}
+
diff --git a/lib/ovsdb-cs.h b/lib/ovsdb-cs.h
index e4af32fd70c9..bbe9588435d2 100644
--- a/lib/ovsdb-cs.h
+++ b/lib/ovsdb-cs.h
@@ -17,14 +17,141 @@
 #ifndef OVSDB_CS_H
 #define OVSDB_CS_H 1
 
-#include "openvswitch/compiler.h"
+/* Open vSwitch Database client synchronization layer.
+ *
+ * This is a base layer for maintaining an in-memory replica of a database.  It
+ * issues RPC requests to an OVSDB database server and passes the semantically
+ * meaningful parts of the stream up to a higher layer.  The OVSDB IDL uses
+ * this as a base layer, as well as OVN's DDlog-based northd implementation.
+ */
+
+#include <stdbool.h>
 #include "openvswitch/hmap.h"
-#include "openvswitch/json.h"
+#include "openvswitch/list.h"
 #include "openvswitch/shash.h"
 #include "openvswitch/uuid.h"
 
-/* Open vSwitch Database client synchronization layer. */
+struct json;
+struct ovsdb_cs;
+
+struct ovsdb_cs_ops {
+    /* Returns <monitor-requests> to use for the specified <schema>.  The
+     * implementation might find ovsdb_cs_parse_table_updates() to be a useful
+     * helper.
+     *
+     * The caller might actually use "monitor_cond" or "monitor_cond_since",
+     * rather than plain "monitor".  If so, this function's implementation
+     * doesn't need to worry about that, because the caller will add the
+     * conditions itself. */
+    struct json *(*compose_monitor_requests)(const struct json *schema,
+                                             void *aux);
+};
+
+/* An event is a happening that is worth reporting to the CS client.
+ *
+ * Currently there are three kinds of events:
+ *
+ *    - "Reconnect": The connection to the database was lost and it is now
+ *      being reconnected.  This means that any transactions submitted by the
+ *      client will never receive a reply (although it's possible that some of
+ *      them were actually committed).  This event has no associated data.
+ *
+ *    - "Locked": The server granted the lock we requested.
+ *
+ *    - "Update": The server sent an update to one or more monitored tables.
+ *      The client can use the associated data to update its idea of the
+ *      snapshot.
+ *
+ *    - "Transaction reply": The server sent a reply to a transaction sent by
+ *      the client using ovsdb_cs_send_transaction().
+ */
+struct ovsdb_cs_event {
+    struct ovs_list list_node;
+
+    enum ovsdb_cs_event_type {
+        OVSDB_CS_EVENT_TYPE_RECONNECT,   /* Connection lost. */
+        OVSDB_CS_EVENT_TYPE_LOCKED,      /* Got the lock we wanted. */
+        OVSDB_CS_EVENT_TYPE_UPDATE,      /* Received update notification. */
+        OVSDB_CS_EVENT_TYPE_TXN_REPLY,   /* Received reply to transaction. */
+    } type;
+
+    union {
+        /* Represents a <table-updates> or <table-updates2> that contains
+         * either the initial data in a monitor reply or a delta received in an
+         * update notification.  The client can use this to update its database
+         * replica.
+         *
+         * If 'clear' is true, then the client should first clear its idea of
+         * what's in the replica before applying the update; otherwise, it's an
+         * incremental update.
+         *
+         * 'table-updates' is a <table-updates> if 'version' if 1, otherwise a
+         * <table-updates2>.  The client can use ovsdb_cs_parse_table_updates()
+         * to parse the update.
+         */
+        struct ovsdb_cs_update_event {
+            bool clear;
+            struct json *table_updates;
+            int version;
+        } update;
 
+        /* The "result" member from a transaction reply.  The transaction is
+         * one sent by the client using ovsdb_cs_send_transaction().  The
+         * client can match 'txn_reply->id' against the ID in a transaction it
+         * sent.  */
+        struct jsonrpc_msg *txn_reply;
+    };
+};
+void ovsdb_cs_event_destroy(struct ovsdb_cs_event *);
+
+/* Lifecycle. */
+struct ovsdb_cs *ovsdb_cs_create(const char *database, int max_version,
+                                 const struct ovsdb_cs_ops *ops,
+                                 void *ops_aux);
+void ovsdb_cs_destroy(struct ovsdb_cs *);
+
+void ovsdb_cs_run(struct ovsdb_cs *, struct ovs_list *events);
+void ovsdb_cs_wait(struct ovsdb_cs *);
+
+/* Network connection. */
+void ovsdb_cs_set_remote(struct ovsdb_cs *, const char *remote, bool retry);
+
+void ovsdb_cs_enable_reconnect(struct ovsdb_cs *);
+void ovsdb_cs_force_reconnect(struct ovsdb_cs *);
+void ovsdb_cs_flag_inconsistency(struct ovsdb_cs *);
+
+bool ovsdb_cs_is_alive(const struct ovsdb_cs *);
+bool ovsdb_cs_is_connected(const struct ovsdb_cs *);
+int ovsdb_cs_get_last_error(const struct ovsdb_cs *);
+
+void ovsdb_cs_set_probe_interval(const struct ovsdb_cs *, int probe_interval);
+
+/* Conditional monitoring (specifying that only rows matching particular
+ * criteria should be monitored).
+ *
+ * Some database servers don't support conditional monitoring; in that case,
+ * the client will get all the rows. */
+unsigned int ovsdb_cs_set_condition(struct ovsdb_cs *, const char *table,
+                                    const struct json *condition);
+unsigned int ovsdb_cs_get_condition_seqno(const struct ovsdb_cs *);
+
+/* Clustered servers. */
+void ovsdb_cs_set_leader_only(struct ovsdb_cs *, bool leader_only);
+void ovsdb_cs_set_shuffle_remotes(struct ovsdb_cs *, bool shuffle);
+void ovsdb_cs_reset_min_index(struct ovsdb_cs *);
+
+/* Database locks. */
+void ovsdb_cs_set_lock(struct ovsdb_cs *, const char *lock_name);
+const char *ovsdb_cs_get_lock(const struct ovsdb_cs *);
+bool ovsdb_cs_has_lock(const struct ovsdb_cs *);
+bool ovsdb_cs_is_lock_contended(const struct ovsdb_cs *);
+
+/* Transactions. */
+bool ovsdb_cs_may_send_transaction(const struct ovsdb_cs *);
+struct json *ovsdb_cs_send_transaction(struct ovsdb_cs *, struct json *ops)
+    OVS_WARN_UNUSED_RESULT;
+bool ovsdb_cs_forget_transaction(struct ovsdb_cs *, const struct json *);
+
 /* Helper for partially parsing the <table-updates> or <table-updates2> that
  * appear in struct ovsdb_cs_update_event.  The helper leaves the data in JSON
  * format, so it doesn't need to know column types. */
@@ -61,6 +188,8 @@ struct ovsdb_error *ovsdb_cs_parse_db_update(
     struct ovsdb_cs_db_update **db_updatep)
     OVS_WARN_UNUSED_RESULT;
 void ovsdb_cs_db_update_destroy(struct ovsdb_cs_db_update *);
+const struct ovsdb_cs_table_update *ovsdb_cs_db_update_find_table(
+    const struct ovsdb_cs_db_update *, const char *table_name);
 
 /* Simple parsing of OVSDB schemas for use by ovsdb_cs clients.  */
 
diff --git a/lib/ovsdb-idl-provider.h b/lib/ovsdb-idl-provider.h
index 00497d940c32..0f38f9b34fdc 100644
--- a/lib/ovsdb-idl-provider.h
+++ b/lib/ovsdb-idl-provider.h
@@ -118,16 +118,10 @@ struct ovsdb_idl_table {
                               * for replication. */
     struct shash columns;    /* Contains "const struct ovsdb_idl_column *"s. */
     struct hmap rows;        /* Contains "struct ovsdb_idl_row"s. */
-    struct ovsdb_idl_db *db; /* Containing db. */
+    struct ovsdb_idl *idl;   /* Containing IDL instance. */
     unsigned int change_seqno[OVSDB_IDL_CHANGE_MAX];
     struct ovs_list indexes;    /* Contains "struct ovsdb_idl_index"s */
     struct ovs_list track_list; /* Tracked rows (ovsdb_idl_row.track_node). */
-    struct ovsdb_idl_condition *ack_cond; /* Last condition acked by the
-                                           * server. */
-    struct ovsdb_idl_condition *req_cond; /* Last condition requested to the
-                                           * server. */
-    struct ovsdb_idl_condition *new_cond; /* Latest condition set by the IDL
-                                           * client. */
 };
 
 struct ovsdb_idl_class {
diff --git a/lib/ovsdb-idl.c b/lib/ovsdb-idl.c
index fdc59c5e7142..61be78da33f2 100644
--- a/lib/ovsdb-idl.c
+++ b/lib/ovsdb-idl.c
@@ -83,204 +83,23 @@ struct ovsdb_idl_arc {
     struct ovsdb_idl_row *dst;  /* Destination row. */
 };
 
-/* Connection state machine.
- *
- * When a JSON-RPC session connects, the IDL sends a "monitor_cond" request for
- * the Database table in the _Server database and transitions to the
- * IDL_S_SERVER_MONITOR_COND_REQUESTED state.  If the session drops and
- * reconnects, or if the IDL receives a "monitor_canceled" notification for a
- * table it is monitoring, the IDL starts over again in the same way. */
-#define OVSDB_IDL_STATES                                                \
-    /* Waits for "get_schema" reply, then sends "monitor_cond"          \
-     * request for the Database table in the _Server database, whose    \
-     * details are informed by the schema, and transitions to           \
-     * IDL_S_SERVER_MONITOR_COND_REQUESTED. */                          \
-    OVSDB_IDL_STATE(SERVER_SCHEMA_REQUESTED)                            \
-                                                                        \
-    /* Waits for "monitor_cond" reply for the Database table:           \
-     *                                                                  \
-     * - If the reply indicates success, and the Database table has a   \
-     *   row for the IDL database:                                      \
-     *                                                                  \
-     *   * If the row indicates that this is a clustered database       \
-     *     that is not connected to the cluster, closes the             \
-     *     connection.  The next connection attempt has a chance at     \
-     *     picking a connected server.                                  \
-     *                                                                  \
-     *   * Otherwise, sends a "monitor_cond_since" request for the IDL  \
-     *     database whose details are informed by the schema            \
-     *     (obtained from the row), and transitions to                  \
-     *     IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED.                     \
-     *                                                                  \
-     * - If the reply indicates success, but the Database table does    \
-     *   not have a row for the IDL database, transitions to            \
-     *   IDL_S_ERROR.                                                   \
-     *                                                                  \
-     * - If the reply indicates failure, sends a "get_schema" request   \
-     *   for the IDL database and transitions to                        \
-     *   IDL_S_DATA_SCHEMA_REQUESTED. */                                \
-    OVSDB_IDL_STATE(SERVER_MONITOR_COND_REQUESTED)                      \
-                                                                        \
-    /* Waits for "get_schema" reply, then sends "monitor_cond"          \
-     * request whose details are informed by the schema, and            \
-     * transitions to IDL_S_DATA_MONITOR_COND_REQUESTED. */             \
-    OVSDB_IDL_STATE(DATA_SCHEMA_REQUESTED)                              \
-                                                                        \
-    /* Waits for "monitor_cond_since" reply.  If successful, replaces   \
-     * the IDL contents by the data carried in the reply and            \
-     * transitions to IDL_S_MONITORING.  On failure, sends a            \
-     * "monitor_cond" request and transitions to                        \
-     * IDL_S_DATA_MONITOR_COND_REQUESTED. */                            \
-    OVSDB_IDL_STATE(DATA_MONITOR_COND_SINCE_REQUESTED)                  \
-                                                                        \
-    /* Waits for "monitor_cond" reply.  If successful, replaces the     \
-     * IDL contents by the data carried in the reply and transitions    \
-     * to IDL_S_MONITORING.  On failure, sends a "monitor" request      \
-     * and transitions to IDL_S_DATA_MONITOR_REQUESTED. */              \
-    OVSDB_IDL_STATE(DATA_MONITOR_COND_REQUESTED)                        \
-                                                                        \
-    /* Waits for "monitor" reply.  If successful, replaces the IDL      \
-     * contents by the data carried in the reply and transitions to     \
-     * IDL_S_MONITORING.  On failure, transitions to IDL_S_ERROR. */    \
-    OVSDB_IDL_STATE(DATA_MONITOR_REQUESTED)                             \
-                                                                        \
-    /* State that processes "update", "update2" or "update3"            \
-     * notifications for the main database (and the Database table      \
-     * in _Server if available).                                        \
-     *                                                                  \
-     * If we're monitoring the Database table and we get notified       \
-     * that the IDL database has been deleted, we close the             \
-     * connection (which will restart the state machine). */            \
-    OVSDB_IDL_STATE(MONITORING)                                         \
-                                                                        \
-    /* Terminal error state that indicates that nothing useful can be   \
-     * done, for example because the database server doesn't actually   \
-     * have the desired database.  We maintain the session with the     \
-     * database server anyway.  If it starts serving the database       \
-     * that we want, or if someone fixes and restarts the database,     \
-     * then it will kill the session and we will automatically          \
-     * reconnect and try again. */                                      \
-    OVSDB_IDL_STATE(ERROR)                                              \
-                                                                        \
-    /* Terminal state that indicates we connected to a useless server   \
-     * in a cluster, e.g. one that is partitioned from the rest of      \
-     * the cluster. We're waiting to retry. */                          \
-    OVSDB_IDL_STATE(RETRY)
-
-enum ovsdb_idl_state {
-#define OVSDB_IDL_STATE(NAME) IDL_S_##NAME,
-    OVSDB_IDL_STATES
-#undef OVSDB_IDL_STATE
-};
-
-static const char *ovsdb_idl_state_to_string(enum ovsdb_idl_state);
-
-enum ovsdb_idl_monitor_method {
-    OVSDB_IDL_MM_MONITOR,
-    OVSDB_IDL_MM_MONITOR_COND,
-    OVSDB_IDL_MM_MONITOR_COND_SINCE
-};
-
-enum ovsdb_idl_monitoring {
-    OVSDB_IDL_NOT_MONITORING,   /* Database is not being monitored. */
-    OVSDB_IDL_MONITORING,       /* Database has "monitor" outstanding. */
-    OVSDB_IDL_MONITORING_COND,  /* Database has "monitor_cond" outstanding. */
-    OVSDB_IDL_MONITORING_COND_SINCE,  /* Database has "monitor_cond_since"
-                                         outstanding. */
-};
-
-struct ovsdb_idl_db {
-    struct ovsdb_idl *idl;
-
-    /* Data. */
+struct ovsdb_idl {
+    struct ovsdb_cs *cs;
     const struct ovsdb_idl_class *class_;
     struct shash table_by_name; /* Contains "struct ovsdb_idl_table *"s.*/
     struct ovsdb_idl_table *tables; /* Array of ->class_->n_tables elements. */
-    struct json *monitor_id;
     unsigned int change_seqno;
     struct ovsdb_idl_txn *txn;
     struct hmap outstanding_txns;
     bool verify_write_only;
-    struct json *schema;
-    enum ovsdb_idl_monitoring monitoring;
-
-    /* True if any of the tables' monitoring conditions has changed. */
-    bool cond_changed;
-
-    unsigned int cond_seqno;   /* Keep track of condition clauses changes
-                                  over a single conditional monitoring session.
-                                  Reverts to zero when idl session
-                                  reconnects.  */
-
-    /* Database locking. */
-    char *lock_name;            /* Name of lock we need, NULL if none. */
-    bool has_lock;              /* Has db server told us we have the lock? */
-    bool is_lock_contended;     /* Has db server told us we can't get lock? */
-    struct json *lock_request_id; /* JSON-RPC ID of in-flight lock request. */
-
-    /* Last db txn id, used for fast resync through monitor_cond_since */
-    struct uuid last_id;
-};
-
-static void ovsdb_idl_db_track_clear(struct ovsdb_idl_db *);
-static void ovsdb_idl_db_add_column(struct ovsdb_idl_db *,
-                                    const struct ovsdb_idl_column *);
-static void ovsdb_idl_db_omit(struct ovsdb_idl_db *,
-                              const struct ovsdb_idl_column *);
-static void ovsdb_idl_db_omit_alert(struct ovsdb_idl_db *,
-                                    const struct ovsdb_idl_column *);
-static unsigned int ovsdb_idl_db_set_condition(
-    struct ovsdb_idl_db *, const struct ovsdb_idl_table_class *,
-    const struct ovsdb_idl_condition *);
-
-static void ovsdb_idl_send_schema_request(struct ovsdb_idl *,
-                                          struct ovsdb_idl_db *);
-static void ovsdb_idl_send_db_change_aware(struct ovsdb_idl *);
-static bool ovsdb_idl_check_server_db(struct ovsdb_idl *);
-static void ovsdb_idl_send_monitor_request(struct ovsdb_idl *,
-                                           struct ovsdb_idl_db *,
-                                           enum ovsdb_idl_monitor_method);
-static void ovsdb_idl_db_clear(struct ovsdb_idl_db *db);
-static void ovsdb_idl_db_ack_condition(struct ovsdb_idl_db *db);
-static void ovsdb_idl_db_sync_condition(struct ovsdb_idl_db *db);
-static void ovsdb_idl_condition_move(struct ovsdb_idl_condition **dst,
-                                     struct ovsdb_idl_condition **src);
-
-struct ovsdb_idl {
-    struct ovsdb_idl_db server;
-    struct ovsdb_idl_db data;
-
-    /* Session state.
-     *
-     *'state_seqno' is a snapshot of the session's sequence number as returned
-     * jsonrpc_session_get_seqno(session), so if it differs from the value that
-     * function currently returns then the session has reconnected and the
-     * state machine must restart.  */
-    struct jsonrpc_session *session; /* Connection to the server. */
-    char *remote;                    /* 'session' remote name. */
-    enum ovsdb_idl_state state;      /* Current session state. */
-    unsigned int state_seqno;        /* See above. */
-    struct json *request_id;         /* JSON ID for request awaiting reply. */
-
-    struct uuid cid;
-
-    uint64_t min_index;
-    bool leader_only;
-    bool shuffle_remotes;
 };
 
-static void ovsdb_idl_transition_at(struct ovsdb_idl *, enum ovsdb_idl_state,
-                                    const char *where);
-#define ovsdb_idl_transition(IDL, STATE) \
-    ovsdb_idl_transition_at(IDL, STATE, OVS_SOURCE_LOCATOR)
-
-static void ovsdb_idl_retry_at(struct ovsdb_idl *, const char *where);
-#define ovsdb_idl_retry(IDL) ovsdb_idl_retry_at(IDL, OVS_SOURCE_LOCATOR)
+static struct ovsdb_cs_ops ovsdb_idl_cs_ops;
 
 struct ovsdb_idl_txn {
     struct hmap_node hmap_node;
     struct json *request_id;
-    struct ovsdb_idl_db *db;
+    struct ovsdb_idl *idl;
     struct hmap txn_rows;
     enum ovsdb_idl_txn_status status;
     char *error;
@@ -310,29 +129,24 @@ static struct vlog_rate_limit syntax_rl = VLOG_RATE_LIMIT_INIT(1, 5);
 static struct vlog_rate_limit semantic_rl = VLOG_RATE_LIMIT_INIT(1, 5);
 static struct vlog_rate_limit other_rl = VLOG_RATE_LIMIT_INIT(1, 5);
 
-static void ovsdb_idl_clear(struct ovsdb_idl *);
-static void ovsdb_idl_db_parse_monitor_reply(struct ovsdb_idl_db *,
-                                             const struct json *result,
-                                             int version);
-static bool ovsdb_idl_db_parse_update_rpc(struct ovsdb_idl_db *,
-                                          const struct jsonrpc_msg *);
-static bool ovsdb_idl_handle_monitor_canceled(struct ovsdb_idl *,
-                                              struct ovsdb_idl_db *,
-                                              const struct jsonrpc_msg *);
-static void ovsdb_idl_db_parse_update(struct ovsdb_idl_db *,
-                                      const struct json *table_updates,
-                                      int version);
 enum update_result {
     OVSDB_IDL_UPDATE_DB_CHANGED,
     OVSDB_IDL_UPDATE_NO_CHANGES,
     OVSDB_IDL_UPDATE_INCONSISTENT,
 };
+static void ovsdb_idl_clear(struct ovsdb_idl *);
 static enum update_result ovsdb_idl_process_update(
     struct ovsdb_idl_table *, const struct ovsdb_cs_row_update *);
-static void ovsdb_idl_insert_row(struct ovsdb_idl_row *, const struct shash *);
+static void ovsdb_idl_insert_row(struct ovsdb_idl_row *,
+                                 const struct shash *values);
 static void ovsdb_idl_delete_row(struct ovsdb_idl_row *);
-static bool ovsdb_idl_modify_row(struct ovsdb_idl_row *, const struct shash *,
-                                 bool xor);
+static bool ovsdb_idl_modify_row(struct ovsdb_idl_row *,
+                                 const struct shash *values, bool xor);
+static void ovsdb_idl_parse_update(struct ovsdb_idl *,
+                                   const struct ovsdb_cs_update_event *);
+
+static void ovsdb_idl_txn_process_reply(struct ovsdb_idl *,
+                                        const struct jsonrpc_msg *);
 
 static bool ovsdb_idl_row_is_orphan(const struct ovsdb_idl_row *);
 static struct ovsdb_idl_row *ovsdb_idl_row_create__(
@@ -340,7 +154,7 @@ static struct ovsdb_idl_row *ovsdb_idl_row_create__(
 static struct ovsdb_idl_row *ovsdb_idl_row_create(struct ovsdb_idl_table *,
                                                   const struct uuid *);
 static void ovsdb_idl_row_destroy(struct ovsdb_idl_row *);
-static void ovsdb_idl_row_destroy_postprocess(struct ovsdb_idl_db *);
+static void ovsdb_idl_row_destroy_postprocess(struct ovsdb_idl *);
 static void ovsdb_idl_destroy_all_map_op_lists(struct ovsdb_idl_row *);
 static void ovsdb_idl_destroy_all_set_op_lists(struct ovsdb_idl_row *);
 
@@ -350,10 +164,7 @@ static void ovsdb_idl_row_clear_old(struct ovsdb_idl_row *);
 static void ovsdb_idl_row_clear_new(struct ovsdb_idl_row *);
 static void ovsdb_idl_row_clear_arcs(struct ovsdb_idl_row *, bool destroy_dsts);
 
-static void ovsdb_idl_db_txn_abort_all(struct ovsdb_idl_db *);
 static void ovsdb_idl_txn_abort_all(struct ovsdb_idl *);
-static bool ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *,
-                                           const struct jsonrpc_msg *msg);
 static bool ovsdb_idl_txn_extract_mutations(struct ovsdb_idl_row *,
                                             struct json *);
 static void ovsdb_idl_txn_add_map_op(struct ovsdb_idl_row *,
@@ -365,25 +176,13 @@ static void ovsdb_idl_txn_add_set_op(struct ovsdb_idl_row *,
                                      struct ovsdb_datum *,
                                      enum set_op_type);
 
-static bool ovsdb_idl_db_process_lock_replies(struct ovsdb_idl_db *,
-                                              const struct jsonrpc_msg *);
-static struct jsonrpc_msg *ovsdb_idl_db_compose_lock_request(
-    struct ovsdb_idl_db *);
-static struct jsonrpc_msg *ovsdb_idl_db_compose_unlock_request(
-    struct ovsdb_idl_db *);
-static void ovsdb_idl_db_parse_lock_reply(struct ovsdb_idl_db *,
-                                          const struct json *);
-static bool ovsdb_idl_db_parse_lock_notify(struct ovsdb_idl_db *,
-                                           const struct json *params,
-                                           bool new_has_lock);
 static struct ovsdb_idl_table *
-ovsdb_idl_db_table_from_class(const struct ovsdb_idl_db *,
+ovsdb_idl_table_from_class(const struct ovsdb_idl *,
                               const struct ovsdb_idl_table_class *);
 static struct ovsdb_idl_table *
 ovsdb_idl_table_from_class(const struct ovsdb_idl *,
                            const struct ovsdb_idl_table_class *);
 static bool ovsdb_idl_track_is_set(struct ovsdb_idl_table *table);
-static void ovsdb_idl_send_cond_change(struct ovsdb_idl *idl);
 
 static void ovsdb_idl_destroy_indexes(struct ovsdb_idl_table *);
 static void ovsdb_idl_add_to_indexes(const struct ovsdb_idl_row *);
@@ -391,53 +190,6 @@ static void ovsdb_idl_remove_from_indexes(const struct ovsdb_idl_row *);
 static int ovsdb_idl_try_commit_loop_txn(struct ovsdb_idl_loop *loop,
                                          bool *may_need_wakeup);
 
-static void
-ovsdb_idl_db_init(struct ovsdb_idl_db *db, const struct ovsdb_idl_class *class,
-                  struct ovsdb_idl *parent, bool monitor_everything_by_default)
-{
-    memset(db, 0, sizeof *db);
-
-    uint8_t default_mode = (monitor_everything_by_default
-                            ? OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT
-                            : 0);
-
-    db->idl = parent;
-    db->class_ = class;
-    shash_init(&db->table_by_name);
-    db->tables = xmalloc(class->n_tables * sizeof *db->tables);
-    for (size_t i = 0; i < class->n_tables; i++) {
-        const struct ovsdb_idl_table_class *tc = &class->tables[i];
-        struct ovsdb_idl_table *table = &db->tables[i];
-
-        shash_add_assert(&db->table_by_name, tc->name, table);
-        table->class_ = tc;
-        table->modes = xmalloc(tc->n_columns);
-        memset(table->modes, default_mode, tc->n_columns);
-        table->need_table = false;
-        shash_init(&table->columns);
-        ovs_list_init(&table->indexes);
-        for (size_t j = 0; j < tc->n_columns; j++) {
-            const struct ovsdb_idl_column *column = &tc->columns[j];
-
-            shash_add_assert(&table->columns, column->name, column);
-        }
-        hmap_init(&table->rows);
-        ovs_list_init(&table->track_list);
-        table->change_seqno[OVSDB_IDL_CHANGE_INSERT]
-            = table->change_seqno[OVSDB_IDL_CHANGE_MODIFY]
-            = table->change_seqno[OVSDB_IDL_CHANGE_DELETE] = 0;
-        table->db = db;
-        table->ack_cond = NULL;
-        table->req_cond = NULL;
-        table->new_cond = xmalloc(sizeof *table->new_cond);
-        ovsdb_idl_condition_init(table->new_cond);
-        ovsdb_idl_condition_add_clause_true(table->new_cond);
-    }
-    db->monitor_id = json_array_create_2(json_string_create("monid"),
-                                         json_string_create(class->database));
-    hmap_init(&db->outstanding_txns);
-}
-
 /* Creates and returns a connection to database 'remote', which should be in a
  * form acceptable to jsonrpc_session_open().  The connection will maintain an
  * in-memory replica of the remote database whose schema is described by
@@ -486,29 +238,44 @@ struct ovsdb_idl *
 ovsdb_idl_create_unconnected(const struct ovsdb_idl_class *class,
                              bool monitor_everything_by_default)
 {
-    struct ovsdb_idl *idl;
+    struct ovsdb_idl *idl = xmalloc(sizeof *idl);
+    *idl = (struct ovsdb_idl) {
+        .cs = ovsdb_cs_create(class->database, 3, &ovsdb_idl_cs_ops, idl),
+        .class_ = class,
+        .table_by_name = SHASH_INITIALIZER(&idl->table_by_name),
+        .tables = xmalloc(class->n_tables * sizeof *idl->tables),
+        .change_seqno = 0,
+        .txn = NULL,
+        .outstanding_txns = HMAP_INITIALIZER(&idl->outstanding_txns),
+        .verify_write_only = false,
+    };
+
+    uint8_t default_mode = (monitor_everything_by_default
+                            ? OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT
+                            : 0);
+    for (size_t i = 0; i < class->n_tables; i++) {
+        const struct ovsdb_idl_table_class *tc = &class->tables[i];
+        struct ovsdb_idl_table *table = &idl->tables[i];
 
-    idl = xzalloc(sizeof *idl);
-    ovsdb_idl_db_init(&idl->server, &serverrec_idl_class, idl, true);
-    ovsdb_idl_db_init(&idl->data, class, idl, monitor_everything_by_default);
-    idl->state_seqno = UINT_MAX;
-    idl->request_id = NULL;
-    idl->leader_only = true;
-    idl->shuffle_remotes = true;
+        shash_add_assert(&idl->table_by_name, tc->name, table);
+        table->class_ = tc;
+        table->modes = xmalloc(tc->n_columns);
+        memset(table->modes, default_mode, tc->n_columns);
+        table->need_table = false;
+        shash_init(&table->columns);
+        ovs_list_init(&table->indexes);
+        for (size_t j = 0; j < tc->n_columns; j++) {
+            const struct ovsdb_idl_column *column = &tc->columns[j];
 
-    /* Monitor the Database table in the _Server database.
-     *
-     * We monitor only the row for 'class', or the row that has the
-     * desired 'cid'. */
-    struct ovsdb_idl_condition cond;
-    ovsdb_idl_condition_init(&cond);
-    if (!uuid_is_zero(&idl->cid)) {
-        serverrec_database_add_clause_cid(&cond, OVSDB_F_EQ, &idl->cid, 1);
-    } else {
-        serverrec_database_add_clause_name(&cond, OVSDB_F_EQ, class->database);
+            shash_add_assert(&table->columns, column->name, column);
+        }
+        hmap_init(&table->rows);
+        ovs_list_init(&table->track_list);
+        table->change_seqno[OVSDB_IDL_CHANGE_INSERT]
+            = table->change_seqno[OVSDB_IDL_CHANGE_MODIFY]
+            = table->change_seqno[OVSDB_IDL_CHANGE_DELETE] = 0;
+        table->idl = idl;
     }
-    ovsdb_idl_db_set_condition(&idl->server, &serverrec_table_database, &cond);
-    ovsdb_idl_condition_destroy(&cond);
 
     return idl;
 }
@@ -520,35 +287,7 @@ ovsdb_idl_create_unconnected(const struct ovsdb_idl_class *class,
 void
 ovsdb_idl_set_remote(struct ovsdb_idl *idl, const char *remote, bool retry)
 {
-    if (idl
-        && ((remote != NULL) != (idl->remote != NULL)
-            || (remote && idl->remote && strcmp(remote, idl->remote)))) {
-        ovs_assert(!idl->data.txn);
-
-        /* Close the old session, if any. */
-        if (idl->session) {
-            jsonrpc_session_close(idl->session);
-            idl->session = NULL;
-
-            free(idl->remote);
-            idl->remote = NULL;
-        }
-
-        /* Open new session, if any. */
-        if (remote) {
-            struct svec remotes = SVEC_EMPTY_INITIALIZER;
-            ovsdb_session_parse_remote(remote, &remotes, &idl->cid);
-            if (idl->shuffle_remotes) {
-                svec_shuffle(&remotes);
-            }
-            idl->session = jsonrpc_session_open_multiple(&remotes, retry);
-            svec_destroy(&remotes);
-
-            idl->state_seqno = UINT_MAX;
-
-            idl->remote = xstrdup(remote);
-        }
-    }
+    ovsdb_cs_set_remote(idl->cs, remote, retry);
 }
 
 /* Set whether the order of remotes should be shuffled, when there
@@ -557,7 +296,7 @@ ovsdb_idl_set_remote(struct ovsdb_idl *idl, const char *remote, bool retry)
 void
 ovsdb_idl_set_shuffle_remotes(struct ovsdb_idl *idl, bool shuffle)
 {
-    idl->shuffle_remotes = shuffle;
+    ovsdb_cs_set_shuffle_remotes(idl->cs, shuffle);
 }
 
 /* Reset min_index to 0. This prevents a situation where the client
@@ -567,33 +306,7 @@ ovsdb_idl_set_shuffle_remotes(struct ovsdb_idl *idl, bool shuffle)
 void
 ovsdb_idl_reset_min_index(struct ovsdb_idl *idl)
 {
-    idl->min_index = 0;
-}
-
-static void
-ovsdb_idl_db_destroy(struct ovsdb_idl_db *db)
-{
-    struct ovsdb_idl_condition *null_cond = NULL;
-    ovs_assert(!db->txn);
-    ovsdb_idl_db_txn_abort_all(db);
-    ovsdb_idl_db_clear(db);
-    for (size_t i = 0; i < db->class_->n_tables; i++) {
-        struct ovsdb_idl_table *table = &db->tables[i];
-        ovsdb_idl_condition_move(&table->ack_cond, &null_cond);
-        ovsdb_idl_condition_move(&table->req_cond, &null_cond);
-        ovsdb_idl_condition_move(&table->new_cond, &null_cond);
-        ovsdb_idl_destroy_indexes(table);
-        shash_destroy(&table->columns);
-        hmap_destroy(&table->rows);
-        free(table->modes);
-    }
-    shash_destroy(&db->table_by_name);
-    free(db->tables);
-    json_destroy(db->schema);
-    hmap_destroy(&db->outstanding_txns);
-    free(db->lock_name);
-    json_destroy(db->lock_request_id);
-    json_destroy(db->monitor_id);
+    ovsdb_cs_reset_min_index(idl->cs);
 }
 
 /* Destroys 'idl' and all of the data structures that it manages. */
@@ -601,12 +314,22 @@ void
 ovsdb_idl_destroy(struct ovsdb_idl *idl)
 {
     if (idl) {
+        ovs_assert(!idl->txn);
+
+        ovsdb_idl_txn_abort_all(idl);
+        hmap_destroy(&idl->outstanding_txns);
+
         ovsdb_idl_clear(idl);
-        jsonrpc_session_close(idl->session);
-        ovsdb_idl_db_destroy(&idl->server);
-        ovsdb_idl_db_destroy(&idl->data);
-        json_destroy(idl->request_id);
-        free(idl->remote);
+        ovsdb_cs_destroy(idl->cs);
+        for (size_t i = 0; i < idl->class_->n_tables; i++) {
+            struct ovsdb_idl_table *table = &idl->tables[i];
+            ovsdb_idl_destroy_indexes(table);
+            shash_destroy(&table->columns);
+            hmap_destroy(&table->rows);
+            free(table->modes);
+        }
+        shash_destroy(&idl->table_by_name);
+        free(idl->tables);
         free(idl);
     }
 }
@@ -622,19 +345,15 @@ ovsdb_idl_destroy(struct ovsdb_idl *idl)
 void
 ovsdb_idl_set_leader_only(struct ovsdb_idl *idl, bool leader_only)
 {
-    idl->leader_only = leader_only;
-    if (leader_only && idl->server.monitoring) {
-        ovsdb_idl_check_server_db(idl);
-    }
+    ovsdb_cs_set_leader_only(idl->cs, leader_only);
 }
 
 static void
-ovsdb_idl_db_clear(struct ovsdb_idl_db *db)
+ovsdb_idl_clear(struct ovsdb_idl *db)
 {
     bool changed = false;
-    size_t i;
 
-    for (i = 0; i < db->class_->n_tables; i++) {
+    for (size_t i = 0; i < db->class_->n_tables; i++) {
         struct ovsdb_idl_table *table = &db->tables[i];
         struct ovsdb_idl_row *row, *next_row;
 
@@ -661,287 +380,49 @@ ovsdb_idl_db_clear(struct ovsdb_idl_db *db)
     }
     ovsdb_idl_row_destroy_postprocess(db);
 
-    db->cond_seqno = 0;
-    ovsdb_idl_db_track_clear(db);
+    ovsdb_idl_track_clear(db);
 
     if (changed) {
         db->change_seqno++;
     }
 }
 
-static const char *
-ovsdb_idl_state_to_string(enum ovsdb_idl_state state)
-{
-    switch (state) {
-#define OVSDB_IDL_STATE(NAME) case IDL_S_##NAME: return #NAME;
-        OVSDB_IDL_STATES
-#undef OVSDB_IDL_STATE
-    default: return "<unknown>";
-    }
-}
-
-static void
-ovsdb_idl_retry_at(struct ovsdb_idl *idl, const char *where)
-{
-    ovsdb_idl_force_reconnect(idl);
-    ovsdb_idl_transition_at(idl, IDL_S_RETRY, where);
-}
-
-static void
-ovsdb_idl_transition_at(struct ovsdb_idl *idl, enum ovsdb_idl_state new_state,
-                        const char *where)
-{
-    VLOG_DBG("%s: %s -> %s at %s",
-             idl->session ? jsonrpc_session_get_name(idl->session) : "void",
-             ovsdb_idl_state_to_string(idl->state),
-             ovsdb_idl_state_to_string(new_state),
-             where);
-    idl->state = new_state;
-}
-
-static void
-ovsdb_idl_clear(struct ovsdb_idl *idl)
-{
-    ovsdb_idl_db_clear(&idl->data);
-}
-
-static void
-ovsdb_idl_send_request(struct ovsdb_idl *idl, struct jsonrpc_msg *request)
-{
-    json_destroy(idl->request_id);
-    idl->request_id = json_clone(request->id);
-    if (idl->session) {
-        jsonrpc_session_send(idl->session, request);
-    } else {
-        jsonrpc_msg_destroy(request);
-    }
-}
-
-static void
-ovsdb_idl_restart_fsm(struct ovsdb_idl *idl)
-{
-    /* Resync data DB table conditions to avoid missing updates due to
-     * conditions that were in flight or changed locally while the connection
-     * was down.
-     */
-    ovsdb_idl_db_sync_condition(&idl->data);
-
-    ovsdb_idl_send_schema_request(idl, &idl->server);
-    ovsdb_idl_transition(idl, IDL_S_SERVER_SCHEMA_REQUESTED);
-    idl->data.monitoring = OVSDB_IDL_NOT_MONITORING;
-    idl->server.monitoring = OVSDB_IDL_NOT_MONITORING;
-}
-
-static void
-ovsdb_idl_process_response(struct ovsdb_idl *idl, struct jsonrpc_msg *msg)
-{
-    bool ok = msg->type == JSONRPC_REPLY;
-    if (!ok
-        && idl->state != IDL_S_SERVER_SCHEMA_REQUESTED
-        && idl->state != IDL_S_SERVER_MONITOR_COND_REQUESTED
-        && idl->state != IDL_S_DATA_MONITOR_COND_REQUESTED
-        && idl->state != IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED) {
-        static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
-        char *s = jsonrpc_msg_to_string(msg);
-        VLOG_INFO_RL(&rl, "%s: received unexpected %s response in "
-                     "%s state: %s", jsonrpc_session_get_name(idl->session),
-                     jsonrpc_msg_type_to_string(msg->type),
-                     ovsdb_idl_state_to_string(idl->state),
-                     s);
-        free(s);
-        ovsdb_idl_retry(idl);
-        return;
-    }
-
-    switch (idl->state) {
-    case IDL_S_SERVER_SCHEMA_REQUESTED:
-        if (ok) {
-            json_destroy(idl->server.schema);
-            idl->server.schema = json_clone(msg->result);
-            ovsdb_idl_send_monitor_request(idl, &idl->server,
-                                           OVSDB_IDL_MM_MONITOR_COND);
-            ovsdb_idl_transition(idl, IDL_S_SERVER_MONITOR_COND_REQUESTED);
-        } else {
-            ovsdb_idl_send_schema_request(idl, &idl->data);
-            ovsdb_idl_transition(idl, IDL_S_DATA_SCHEMA_REQUESTED);
-        }
-        break;
-
-    case IDL_S_SERVER_MONITOR_COND_REQUESTED:
-        if (ok) {
-            idl->server.monitoring = OVSDB_IDL_MONITORING_COND;
-            ovsdb_idl_db_parse_monitor_reply(&idl->server, msg->result, 2);
-            if (ovsdb_idl_check_server_db(idl)) {
-                ovsdb_idl_send_db_change_aware(idl);
-            }
-        } else {
-            ovsdb_idl_send_schema_request(idl, &idl->data);
-            ovsdb_idl_transition(idl, IDL_S_DATA_SCHEMA_REQUESTED);
-        }
-        break;
-
-    case IDL_S_DATA_SCHEMA_REQUESTED:
-        json_destroy(idl->data.schema);
-        idl->data.schema = json_clone(msg->result);
-        ovsdb_idl_send_monitor_request(idl, &idl->data,
-                                       OVSDB_IDL_MM_MONITOR_COND);
-        ovsdb_idl_transition(idl, IDL_S_DATA_MONITOR_COND_REQUESTED);
-        break;
-
-    case IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED:
-        if (!ok) {
-            /* "monitor_cond_since" not supported.  Try "monitor_cond". */
-            ovsdb_idl_send_monitor_request(idl, &idl->data,
-                                           OVSDB_IDL_MM_MONITOR_COND);
-            ovsdb_idl_transition(idl, IDL_S_DATA_MONITOR_COND_REQUESTED);
-        } else {
-            idl->data.monitoring = OVSDB_IDL_MONITORING_COND_SINCE;
-            ovsdb_idl_transition(idl, IDL_S_MONITORING);
-            ovsdb_idl_db_parse_monitor_reply(&idl->data, msg->result, 3);
-        }
-        break;
-
-    case IDL_S_DATA_MONITOR_COND_REQUESTED:
-        if (!ok) {
-            /* "monitor_cond" not supported.  Try "monitor". */
-            ovsdb_idl_send_monitor_request(idl, &idl->data,
-                                           OVSDB_IDL_MM_MONITOR);
-            ovsdb_idl_transition(idl, IDL_S_DATA_MONITOR_REQUESTED);
-        } else {
-            idl->data.monitoring = OVSDB_IDL_MONITORING_COND;
-            ovsdb_idl_transition(idl, IDL_S_MONITORING);
-            ovsdb_idl_db_parse_monitor_reply(&idl->data, msg->result, 2);
-        }
-        break;
-
-    case IDL_S_DATA_MONITOR_REQUESTED:
-        idl->data.monitoring = OVSDB_IDL_MONITORING;
-        ovsdb_idl_transition(idl, IDL_S_MONITORING);
-        ovsdb_idl_db_parse_monitor_reply(&idl->data, msg->result, 1);
-        idl->data.change_seqno++;
-        break;
-
-    case IDL_S_MONITORING:
-        /* We don't normally have a request outstanding in this state.  If we
-         * do, it's a "monitor_cond_change", which means that the conditional
-         * monitor clauses were updated.
-         *
-         * Mark the last requested conditions as acked and if further
-         * condition changes were pending, send them now. */
-        ovsdb_idl_db_ack_condition(&idl->data);
-        ovsdb_idl_send_cond_change(idl);
-        idl->data.cond_seqno++;
-        break;
-
-    case IDL_S_ERROR:
-    case IDL_S_RETRY:
-        /* Nothing to do in this state. */
-        break;
-
-    default:
-        OVS_NOT_REACHED();
-    }
-}
-
-static void
-ovsdb_idl_process_msg(struct ovsdb_idl *idl, struct jsonrpc_msg *msg)
-{
-    bool is_response = (msg->type == JSONRPC_REPLY ||
-                        msg->type == JSONRPC_ERROR);
-
-    /* Process a reply to an outstanding request. */
-    if (is_response
-        && idl->request_id && json_equal(idl->request_id, msg->id)) {
-        json_destroy(idl->request_id);
-        idl->request_id = NULL;
-        ovsdb_idl_process_response(idl, msg);
-        return;
-    }
-
-    /* Process database contents updates. */
-    if (ovsdb_idl_db_parse_update_rpc(&idl->data, msg)) {
-        return;
-    }
-    if (idl->server.monitoring
-        && ovsdb_idl_db_parse_update_rpc(&idl->server, msg)) {
-        ovsdb_idl_check_server_db(idl);
-        return;
-    }
-
-    if (ovsdb_idl_handle_monitor_canceled(idl, &idl->data, msg)
-        || (idl->server.monitoring
-            && ovsdb_idl_handle_monitor_canceled(idl, &idl->server, msg))) {
-        return;
-    }
-
-    /* Process "lock" replies and related notifications. */
-    if (ovsdb_idl_db_process_lock_replies(&idl->data, msg)) {
-        return;
-    }
-
-    /* Process response to a database transaction we submitted. */
-    if (is_response && ovsdb_idl_db_txn_process_reply(&idl->data, msg)) {
-        return;
-    }
-
-    /* Unknown message.  Log at a low level because this can happen if
-     * ovsdb_idl_txn_destroy() is called to destroy a transaction
-     * before we receive the reply.
-     *
-     * (We could sort those out from other kinds of unknown messages by
-     * using distinctive IDs for transactions, if it seems valuable to
-     * do so, and then it would be possible to use different log
-     * levels. XXX?) */
-    char *s = jsonrpc_msg_to_string(msg);
-    VLOG_DBG("%s: received unexpected %s message: %s",
-             jsonrpc_session_get_name(idl->session),
-             jsonrpc_msg_type_to_string(msg->type), s);
-    free(s);
-}
-
 /* Processes a batch of messages from the database server on 'idl'.  This may
  * cause the IDL's contents to change.  The client may check for that with
  * ovsdb_idl_get_seqno(). */
 void
 ovsdb_idl_run(struct ovsdb_idl *idl)
 {
-    if (!idl->session) {
-        ovsdb_idl_txn_abort_all(idl);
-        return;
-    }
-
-    int i;
-
-    ovs_assert(!idl->data.txn);
+    ovs_assert(!idl->txn);
 
-    ovsdb_idl_send_cond_change(idl);
+    struct ovs_list events;
+    ovsdb_cs_run(idl->cs, &events);
 
-    jsonrpc_session_run(idl->session);
-    for (i = 0; jsonrpc_session_is_connected(idl->session) && i < 50; i++) {
-        struct jsonrpc_msg *msg;
-        unsigned int seqno;
-
-        seqno = jsonrpc_session_get_seqno(idl->session);
-        if (idl->state_seqno != seqno) {
-            idl->state_seqno = seqno;
+    struct ovsdb_cs_event *event;
+    LIST_FOR_EACH_POP (event, list_node, &events) {
+        switch (event->type) {
+        case OVSDB_CS_EVENT_TYPE_RECONNECT:
             ovsdb_idl_txn_abort_all(idl);
-            ovsdb_idl_restart_fsm(idl);
+            break;
 
-            if (idl->data.lock_name) {
-                jsonrpc_session_send(
-                    idl->session,
-                    ovsdb_idl_db_compose_lock_request(&idl->data));
-            }
-        }
+        case OVSDB_CS_EVENT_TYPE_LOCKED:
+            /* If the client couldn't run a transaction because it didn't have
+             * the lock, this will encourage it to try again. */
+            idl->change_seqno++;
+            break;
+
+        case OVSDB_CS_EVENT_TYPE_UPDATE:
+            ovsdb_idl_parse_update(idl, &event->update);
+            break;
 
-        msg = jsonrpc_session_recv(idl->session);
-        if (!msg) {
+        case OVSDB_CS_EVENT_TYPE_TXN_REPLY:
+            ovsdb_idl_txn_process_reply(idl, event->txn_reply);
             break;
         }
-        ovsdb_idl_process_msg(idl, msg);
-        jsonrpc_msg_destroy(msg);
+        ovsdb_cs_event_destroy(event);
     }
-    ovsdb_idl_row_destroy_postprocess(&idl->data);
+
+    ovsdb_idl_row_destroy_postprocess(idl);
 }
 
 /* Arranges for poll_block() to wake up when ovsdb_idl_run() has something to
@@ -949,11 +430,7 @@ ovsdb_idl_run(struct ovsdb_idl *idl)
 void
 ovsdb_idl_wait(struct ovsdb_idl *idl)
 {
-    if (!idl->session) {
-        return;
-    }
-    jsonrpc_session_wait(idl->session);
-    jsonrpc_session_recv_wait(idl->session);
+    ovsdb_cs_wait(idl->cs);
 }
 
 /* Returns a "sequence number" that represents the state of 'idl'.  When
@@ -976,7 +453,7 @@ ovsdb_idl_wait(struct ovsdb_idl *idl)
 unsigned int
 ovsdb_idl_get_seqno(const struct ovsdb_idl *idl)
 {
-    return idl->data.change_seqno;
+    return idl->change_seqno;
 }
 
 /* Returns a "sequence number" that represents the number of conditional
@@ -996,7 +473,7 @@ ovsdb_idl_get_seqno(const struct ovsdb_idl *idl)
 unsigned int
 ovsdb_idl_get_condition_seqno(const struct ovsdb_idl *idl)
 {
-    return idl->data.cond_seqno;
+    return ovsdb_cs_get_condition_seqno(idl->cs);
 }
 
 /* Returns true if 'idl' successfully connected to the remote database and
@@ -1018,9 +495,7 @@ ovsdb_idl_has_ever_connected(const struct ovsdb_idl *idl)
 void
 ovsdb_idl_enable_reconnect(struct ovsdb_idl *idl)
 {
-    if (idl->session) {
-        jsonrpc_session_enable_reconnect(idl->session);
-    }
+    ovsdb_cs_enable_reconnect(idl->cs);
 }
 
 /* Forces 'idl' to drop its connection to the database and reconnect.  In the
@@ -1028,9 +503,7 @@ ovsdb_idl_enable_reconnect(struct ovsdb_idl *idl)
 void
 ovsdb_idl_force_reconnect(struct ovsdb_idl *idl)
 {
-    if (idl->session) {
-        jsonrpc_session_force_reconnect(idl->session);
-    }
+    ovsdb_cs_force_reconnect(idl->cs);
 }
 
 /* Some IDL users should only write to write-only columns.  Furthermore,
@@ -1040,7 +513,7 @@ ovsdb_idl_force_reconnect(struct ovsdb_idl *idl)
 void
 ovsdb_idl_verify_write_only(struct ovsdb_idl *idl)
 {
-    idl->data.verify_write_only = true;
+    idl->verify_write_only = true;
 }
 
 /* Returns true if 'idl' is currently connected or trying to connect
@@ -1048,14 +521,13 @@ ovsdb_idl_verify_write_only(struct ovsdb_idl *idl)
 bool
 ovsdb_idl_is_alive(const struct ovsdb_idl *idl)
 {
-    return idl->session && jsonrpc_session_is_alive(idl->session) &&
-           idl->state != IDL_S_ERROR;
+    return ovsdb_cs_is_alive(idl->cs);
 }
 
 bool
 ovsdb_idl_is_connected(const struct ovsdb_idl *idl)
 {
-    return idl->session && jsonrpc_session_is_connected(idl->session);
+    return ovsdb_cs_is_connected(idl->cs);
 }
 
 /* Returns the last error reported on a connection by 'idl'.  The return value
@@ -1066,14 +538,7 @@ ovsdb_idl_is_connected(const struct ovsdb_idl *idl)
 int
 ovsdb_idl_get_last_error(const struct ovsdb_idl *idl)
 {
-    int err = idl->session ? jsonrpc_session_get_last_error(idl->session) : 0;
-    if (err) {
-        return err;
-    } else if (idl->state == IDL_S_ERROR) {
-        return ENOENT;
-    } else {
-        return 0;
-    }
+    return ovsdb_cs_get_last_error(idl->cs);
 }
 
 /* Sets the "probe interval" for 'idl->session' to 'probe_interval', in
@@ -1082,9 +547,7 @@ ovsdb_idl_get_last_error(const struct ovsdb_idl *idl)
 void
 ovsdb_idl_set_probe_interval(const struct ovsdb_idl *idl, int probe_interval)
 {
-    if (idl->session) {
-        jsonrpc_session_set_probe_interval(idl->session, probe_interval);
-    }
+    ovsdb_cs_set_probe_interval(idl->cs, probe_interval);
 }
 
 static size_t
@@ -1155,7 +618,7 @@ void
 ovsdb_idl_check_consistency(const struct ovsdb_idl *idl)
 {
     /* Consistency is broken while a transaction is in progress. */
-    if (!idl->data.txn) {
+    if (!idl->txn) {
         return;
     }
 
@@ -1164,8 +627,8 @@ ovsdb_idl_check_consistency(const struct ovsdb_idl *idl)
     struct uuid *dsts = NULL;
     size_t allocated_dsts = 0;
 
-    for (size_t i = 0; i < idl->data.class_->n_tables; i++) {
-        const struct ovsdb_idl_table *table = &idl->data.tables[i];
+    for (size_t i = 0; i < idl->class_->n_tables; i++) {
+        const struct ovsdb_idl_table *table = &idl->tables[i];
         const struct ovsdb_idl_table_class *class = table->class_;
 
         const struct ovsdb_idl_row *row;
@@ -1208,88 +671,145 @@ ovsdb_idl_check_consistency(const struct ovsdb_idl *idl)
     free(dsts);
     ovs_assert(ok);
 }
-
-const struct ovsdb_idl_class *
-ovsdb_idl_get_class(const struct ovsdb_idl *idl)
-{
-    return idl->data.class_;
-}
-
-/* Given 'column' in some table in 'class', returns the table's class. */
-const struct ovsdb_idl_table_class *
-ovsdb_idl_table_class_from_column(const struct ovsdb_idl_class *class,
-                                  const struct ovsdb_idl_column *column)
-{
-    for (size_t i = 0; i < class->n_tables; i++) {
-        const struct ovsdb_idl_table_class *tc = &class->tables[i];
-        if (column >= tc->columns && column < &tc->columns[tc->n_columns]) {
-            return tc;
-        }
-    }
-
-    OVS_NOT_REACHED();
-}
 
-/* Given 'column' in some table in 'db', returns the table. */
-static struct ovsdb_idl_table *
-ovsdb_idl_table_from_column(struct ovsdb_idl_db *db,
-                            const struct ovsdb_idl_column *column)
-{
-    const struct ovsdb_idl_table_class *tc =
-        ovsdb_idl_table_class_from_column(db->class_, column);
-    return &db->tables[tc - db->class_->tables];
-}
-
-static unsigned char *
-ovsdb_idl_db_get_mode(struct ovsdb_idl_db *db,
-                      const struct ovsdb_idl_column *column)
+static struct json *
+ovsdb_idl_compose_monitor_request(const struct json *schema_json, void *idl_)
 {
-    ovs_assert(!db->change_seqno);
+    struct ovsdb_idl *idl = idl_;
 
-    const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(db,
-                                                                      column);
-    return &table->modes[column - table->class_->columns];
-}
+    struct shash *schema = ovsdb_cs_parse_schema(schema_json);
+    struct json *monitor_requests = json_object_create();
 
-static void
-ovsdb_idl_db_set_mode(struct ovsdb_idl_db *db,
-                      const struct ovsdb_idl_column *column,
-                      unsigned char mode)
-{
-    const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(db,
-                                                                      column);
-    size_t column_idx = column - table->class_->columns;
+    for (size_t i = 0; i < idl->class_->n_tables; i++) {
+        struct ovsdb_idl_table *table = &idl->tables[i];
+        const struct ovsdb_idl_table_class *tc = table->class_;
+        struct json *monitor_request;
+        const struct sset *table_schema
+            = schema ? shash_find_data(schema, table->class_->name) : NULL;
 
-    if (table->modes[column_idx] != mode) {
-        *ovsdb_idl_db_get_mode(db, column) = mode;
-    }
-}
+        struct json *columns
+            = table->need_table ? json_array_create_empty() : NULL;
+        for (size_t j = 0; j < tc->n_columns; j++) {
+            const struct ovsdb_idl_column *column = &tc->columns[j];
+            bool idl_has_column = (table_schema &&
+                                  sset_contains(table_schema, column->name));
+            if (column->is_synthetic) {
+                if (idl_has_column) {
+                    VLOG_WARN("%s table in %s database has synthetic "
+                              "column %s", table->class_->name,
+                              idl->class_->database, column->name);
+                }
+            } else if (table->modes[j] & OVSDB_IDL_MONITOR) {
+                if (table_schema && !idl_has_column) {
+                    VLOG_WARN("%s table in %s database lacks %s column "
+                              "(database needs upgrade?)",
+                              table->class_->name, idl->class_->database,
+                              column->name);
+                    continue;
+                }
+                if (!columns) {
+                    columns = json_array_create_empty();
+                }
+                json_array_add(columns, json_string_create(column->name));
+            }
+        }
+
+        if (columns) {
+            if (schema && !table_schema) {
+                VLOG_WARN("%s database lacks %s table "
+                          "(database needs upgrade?)",
+                          idl->class_->database, table->class_->name);
+                json_destroy(columns);
+                continue;
+            }
+
+            monitor_request = json_object_create();
+            json_object_put(monitor_request, "columns", columns);
+            json_object_put(monitor_requests, tc->name,
+                            json_array_create_1(monitor_request));
+        }
+    }
+    ovsdb_cs_free_schema(schema);
+
+    return monitor_requests;
+}
+
+static struct ovsdb_cs_ops ovsdb_idl_cs_ops = {
+    ovsdb_idl_compose_monitor_request,
+};
+
+const struct ovsdb_idl_class *
+ovsdb_idl_get_class(const struct ovsdb_idl *idl)
+{
+    return idl->class_;
+}
+
+/* Given 'column' in some table in 'class', returns the table's class. */
+const struct ovsdb_idl_table_class *
+ovsdb_idl_table_class_from_column(const struct ovsdb_idl_class *class,
+                                  const struct ovsdb_idl_column *column)
+{
+    for (size_t i = 0; i < class->n_tables; i++) {
+        const struct ovsdb_idl_table_class *tc = &class->tables[i];
+        if (column >= tc->columns && column < &tc->columns[tc->n_columns]) {
+            return tc;
+        }
+    }
+
+    OVS_NOT_REACHED();
+}
+
+/* Given 'column' in some table in 'idl', returns the table. */
+static struct ovsdb_idl_table *
+ovsdb_idl_table_from_column(struct ovsdb_idl *idl,
+                            const struct ovsdb_idl_column *column)
+{
+    const struct ovsdb_idl_table_class *tc =
+        ovsdb_idl_table_class_from_column(idl->class_, column);
+    return &idl->tables[tc - idl->class_->tables];
+}
+
+static unsigned char *
+ovsdb_idl_get_mode(struct ovsdb_idl *idl,
+                   const struct ovsdb_idl_column *column)
+{
+    ovs_assert(!idl->change_seqno);
+
+    const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(idl,
+                                                                      column);
+    return &table->modes[column - table->class_->columns];
+}
+
+static void
+ovsdb_idl_set_mode(struct ovsdb_idl *idl,
+                   const struct ovsdb_idl_column *column,
+                   unsigned char mode)
+{
+    const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(idl,
+                                                                      column);
+    size_t column_idx = column - table->class_->columns;
+
+    if (table->modes[column_idx] != mode) {
+        *ovsdb_idl_get_mode(idl, column) = mode;
+    }
+}
 
 static void
-add_ref_table(struct ovsdb_idl_db *db, const struct ovsdb_base_type *base)
+add_ref_table(struct ovsdb_idl *idl, const struct ovsdb_base_type *base)
 {
     if (base->type == OVSDB_TYPE_UUID && base->uuid.refTableName) {
         struct ovsdb_idl_table *table;
 
-        table = shash_find_data(&db->table_by_name, base->uuid.refTableName);
+        table = shash_find_data(&idl->table_by_name, base->uuid.refTableName);
         if (table) {
             table->need_table = true;
         } else {
             VLOG_WARN("%s IDL class missing referenced table %s",
-                      db->class_->database, base->uuid.refTableName);
+                      idl->class_->database, base->uuid.refTableName);
         }
     }
 }
 
-static void
-ovsdb_idl_db_add_column(struct ovsdb_idl_db *db,
-                        const struct ovsdb_idl_column *column)
-{
-    ovsdb_idl_db_set_mode(db, column, OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT);
-    add_ref_table(db, &column->type.key);
-    add_ref_table(db, &column->type.value);
-}
-
 /* Turns on OVSDB_IDL_MONITOR and OVSDB_IDL_ALERT for 'column' in 'idl'.  Also
  * ensures that any tables referenced by 'column' will be replicated, even if
  * no columns in that table are selected for replication (see
@@ -1303,25 +823,9 @@ void
 ovsdb_idl_add_column(struct ovsdb_idl *idl,
                      const struct ovsdb_idl_column *column)
 {
-    ovsdb_idl_db_add_column(&idl->data, column);
-}
-
-static void
-ovsdb_idl_db_add_table(struct ovsdb_idl_db *db,
-                       const struct ovsdb_idl_table_class *tc)
-{
-    size_t i;
-
-    for (i = 0; i < db->class_->n_tables; i++) {
-        struct ovsdb_idl_table *table = &db->tables[i];
-
-        if (table->class_ == tc) {
-            table->need_table = true;
-            return;
-        }
-    }
-
-    OVS_NOT_REACHED();
+    ovsdb_idl_set_mode(idl, column, OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT);
+    add_ref_table(idl, &column->type.key);
+    add_ref_table(idl, &column->type.value);
 }
 
 /* Ensures that the table with class 'tc' will be replicated on 'idl' even if
@@ -1341,7 +845,16 @@ void
 ovsdb_idl_add_table(struct ovsdb_idl *idl,
                     const struct ovsdb_idl_table_class *tc)
 {
-    ovsdb_idl_db_add_table(&idl->data, tc);
+    for (size_t i = 0; i < idl->class_->n_tables; i++) {
+        struct ovsdb_idl_table *table = &idl->tables[i];
+
+        if (table->class_ == tc) {
+            table->need_table = true;
+            return;
+        }
+    }
+
+    OVS_NOT_REACHED();
 }
 
 /* A single clause within an ovsdb_idl_condition. */
@@ -1502,111 +1015,11 @@ ovsdb_idl_condition_add_clause_true(struct ovsdb_idl_condition *condition)
     }
 }
 
-static bool
-ovsdb_idl_condition_equals(const struct ovsdb_idl_condition *a,
-                           const struct ovsdb_idl_condition *b)
-{
-    if (hmap_count(&a->clauses) != hmap_count(&b->clauses)) {
-        return false;
-    }
-    if (a->is_true != b->is_true) {
-        return false;
-    }
-
-    const struct ovsdb_idl_clause *clause;
-    HMAP_FOR_EACH (clause, hmap_node, &a->clauses) {
-        if (!ovsdb_idl_condition_find_clause(b, clause,
-                                             clause->hmap_node.hash)) {
-            return false;
-        }
-    }
-    return true;
-}
-
-static void
-ovsdb_idl_condition_clone(struct ovsdb_idl_condition **dst,
-                          const struct ovsdb_idl_condition *src)
-{
-    if (*dst) {
-        ovsdb_idl_condition_destroy(*dst);
-    } else {
-        *dst = xmalloc(sizeof **dst);
-    }
-    ovsdb_idl_condition_init(*dst);
-
-    (*dst)->is_true = src->is_true;
-
-    const struct ovsdb_idl_clause *clause;
-    HMAP_FOR_EACH (clause, hmap_node, &src->clauses) {
-        ovsdb_idl_condition_add_clause__(*dst, clause, clause->hmap_node.hash);
-    }
-}
-
-static void
-ovsdb_idl_condition_move(struct ovsdb_idl_condition **dst,
-                         struct ovsdb_idl_condition **src)
-{
-    if (*dst) {
-        ovsdb_idl_condition_destroy(*dst);
-        free(*dst);
-    }
-    *dst = *src;
-    *src = NULL;
-}
-
-static unsigned int
-ovsdb_idl_db_set_condition(struct ovsdb_idl_db *db,
-                           const struct ovsdb_idl_table_class *tc,
-                           const struct ovsdb_idl_condition *condition)
-{
-    struct ovsdb_idl_condition *table_cond;
-    struct ovsdb_idl_table *table = ovsdb_idl_db_table_from_class(db, tc);
-
-    /* Compare the new condition to the last known condition which can be
-     * either "new" (not sent yet), "requested" or "acked", in this order.
-     */
-    if (table->new_cond) {
-        table_cond = table->new_cond;
-    } else if (table->req_cond) {
-        table_cond = table->req_cond;
-    } else {
-        table_cond = table->ack_cond;
-    }
-    ovs_assert(table_cond);
-
-    if (!ovsdb_idl_condition_equals(condition, table_cond)) {
-        ovsdb_idl_condition_clone(&table->new_cond, condition);
-        db->cond_changed = true;
-        poll_immediate_wake();
-        return db->cond_seqno + 1;
-    } else if (table_cond != table->ack_cond) {
-        /* 'condition' was already set but has not been "acked" yet.  The IDL
-         * will be up to date when db->cond_seqno gets incremented. */
-        return db->cond_seqno + 1;
-    }
-
-    return db->cond_seqno;
-}
-
-/* Sets the replication condition for 'tc' in 'idl' to 'condition' and
- * arranges to send the new condition to the database server.
- *
- * Return the next conditional update sequence number.  When this
- * value and ovsdb_idl_get_condition_seqno() matches, the 'idl'
- * contains rows that match the 'condition'. */
-unsigned int
-ovsdb_idl_set_condition(struct ovsdb_idl *idl,
-                        const struct ovsdb_idl_table_class *tc,
-                        const struct ovsdb_idl_condition *condition)
-{
-    return ovsdb_idl_db_set_condition(&idl->data, tc, condition);
-}
-
 static struct json *
 ovsdb_idl_condition_to_json(const struct ovsdb_idl_condition *cnd)
 {
     if (cnd->is_true) {
-        return json_array_create_empty();
+        return NULL;
     }
 
     size_t n = hmap_count(&cnd->clauses);
@@ -1623,158 +1036,22 @@ ovsdb_idl_condition_to_json(const struct ovsdb_idl_condition *cnd)
     ovs_assert(i == n);
     return json_array_create(clauses, n);
 }
-
-static struct json *
-ovsdb_idl_create_cond_change_req(const struct ovsdb_idl_condition *cond)
-{
-    struct json *monitor_cond_change_request = json_object_create();
-    struct json *cond_json = ovsdb_idl_condition_to_json(cond);
-
-    json_object_put(monitor_cond_change_request, "where", cond_json);
-
-    return monitor_cond_change_request;
-}
-
-static struct jsonrpc_msg *
-ovsdb_idl_db_compose_cond_change(struct ovsdb_idl_db *db)
-{
-    if (!db->cond_changed) {
-        return NULL;
-    }
-
-    struct json *monitor_cond_change_requests = NULL;
-    for (size_t i = 0; i < db->class_->n_tables; i++) {
-        struct ovsdb_idl_table *table = &db->tables[i];
-
-        /* Always use the most recent conditions set by the IDL client when
-         * requesting monitor_cond_change, i.e., table->new_cond.
-         */
-        if (table->new_cond) {
-            struct json *req =
-                ovsdb_idl_create_cond_change_req(table->new_cond);
-            if (req) {
-                if (!monitor_cond_change_requests) {
-                    monitor_cond_change_requests = json_object_create();
-                }
-                json_object_put(monitor_cond_change_requests,
-                             table->class_->name,
-                             json_array_create_1(req));
-            }
-            /* Mark the new condition as requested by moving it to req_cond.
-             * If there's already requested condition that's a bug.
-             */
-            ovs_assert(table->req_cond == NULL);
-            ovsdb_idl_condition_move(&table->req_cond, &table->new_cond);
-        }
-    }
-
-    if (!monitor_cond_change_requests) {
-        return NULL;
-    }
-
-    db->cond_changed = false;
-    struct json *params = json_array_create_3(json_clone(db->monitor_id),
-                                              json_clone(db->monitor_id),
-                                              monitor_cond_change_requests);
-    return jsonrpc_create_request("monitor_cond_change", params, NULL);
-}
-
-/* Marks all requested table conditions in 'db' as acked by the server.
- * It should be called when the server replies to monitor_cond_change
- * requests.
- */
-static void
-ovsdb_idl_db_ack_condition(struct ovsdb_idl_db *db)
-{
-    for (size_t i = 0; i < db->class_->n_tables; i++) {
-        struct ovsdb_idl_table *table = &db->tables[i];
-
-        if (table->req_cond) {
-            ovsdb_idl_condition_move(&table->ack_cond, &table->req_cond);
-        }
-    }
-}
-
-/* Should be called when the IDL fsm is restarted and resyncs table conditions
- * based on the state the DB is in:
- * - if a non-zero last_id is available for the DB then upon reconnect
- *   the IDL should first request acked conditions to avoid missing updates
- *   about records that were added before the transaction with
- *   txn-id == last_id. If there were requested condition changes in flight
- *   (i.e., req_cond not NULL) and the IDL client didn't set new conditions
- *   (i.e., new_cond is NULL) then move req_cond to new_cond to trigger a
- *   follow up monitor_cond_change request.
- * - if there's no last_id available for the DB then it's safe to use the
- *   latest conditions set by the IDL client even if they weren't acked yet.
- */
-static void
-ovsdb_idl_db_sync_condition(struct ovsdb_idl_db *db)
-{
-    bool ack_all = uuid_is_zero(&db->last_id);
-
-    db->cond_changed = false;
-    for (size_t i = 0; i < db->class_->n_tables; i++) {
-        struct ovsdb_idl_table *table = &db->tables[i];
-
-        /* When monitor_cond_since requests will be issued, the
-         * table->ack_cond condition will be added to the "where" clause".
-         * Follow up monitor_cond_change requests will use table->new_cond.
-         */
-        if (ack_all) {
-            if (table->new_cond) {
-                ovsdb_idl_condition_move(&table->req_cond, &table->new_cond);
-            }
-
-            if (table->req_cond) {
-                ovsdb_idl_condition_move(&table->ack_cond, &table->req_cond);
-            }
-        } else {
-            /* If there was no "unsent" condition but instead a
-             * monitor_cond_change request was in flight, move table->req_cond
-             * to table->new_cond and set db->cond_changed to trigger a new
-             * monitor_cond_change request.
-             *
-             * However, if a new condition has been set by the IDL client,
-             * monitor_cond_change will be sent anyway and will use the most
-             * recent table->new_cond so there's no need to update it here.
-             */
-            if (table->req_cond && !table->new_cond) {
-                ovsdb_idl_condition_move(&table->new_cond, &table->req_cond);
-                db->cond_changed = true;
-            }
-        }
-    }
-}
-
-static void
-ovsdb_idl_send_cond_change(struct ovsdb_idl *idl)
-{
-    /* When 'idl->request_id' is not NULL, there is an outstanding
-     * conditional monitoring update request that we have not heard
-     * from the server yet. Don't generate another request in this case. */
-    if (!jsonrpc_session_is_connected(idl->session)
-        || idl->data.monitoring == OVSDB_IDL_MONITORING
-        || idl->request_id) {
-        return;
-    }
 
-    struct jsonrpc_msg *msg = ovsdb_idl_db_compose_cond_change(&idl->data);
-    if (msg) {
-        idl->request_id = json_clone(msg->id);
-        jsonrpc_session_send(idl->session, msg);
-    }
-}
-
-/* Turns off OVSDB_IDL_ALERT and OVSDB_IDL_TRACK for 'column' in 'db'.
+/* Sets the replication condition for 'tc' in 'idl' to 'condition' and
+ * arranges to send the new condition to the database server.
  *
- * This function should be called between ovsdb_idl_create() and the first call
- * to ovsdb_idl_run().
- */
-static void
-ovsdb_idl_db_omit_alert(struct ovsdb_idl_db *db,
-                        const struct ovsdb_idl_column *column)
+ * Return the next conditional update sequence number.  When this
+ * value and ovsdb_idl_get_condition_seqno() matches, the 'idl'
+ * contains rows that match the 'condition'. */
+unsigned int
+ovsdb_idl_set_condition(struct ovsdb_idl *idl,
+                        const struct ovsdb_idl_table_class *tc,
+                        const struct ovsdb_idl_condition *condition)
 {
-    *ovsdb_idl_db_get_mode(db, column) &= ~(OVSDB_IDL_ALERT | OVSDB_IDL_TRACK);
+    struct json *cond_json = ovsdb_idl_condition_to_json(condition);
+    unsigned int seqno = ovsdb_cs_set_condition(idl->cs, tc->name, cond_json);
+    json_destroy(cond_json);
+    return seqno;
 }
 
 /* Turns off OVSDB_IDL_ALERT and OVSDB_IDL_TRACK for 'column' in 'idl'.
@@ -1786,14 +1063,7 @@ void
 ovsdb_idl_omit_alert(struct ovsdb_idl *idl,
                      const struct ovsdb_idl_column *column)
 {
-    ovsdb_idl_db_omit_alert(&idl->data, column);
-}
-
-static void
-ovsdb_idl_db_omit(struct ovsdb_idl_db *db,
-                  const struct ovsdb_idl_column *column)
-{
-    *ovsdb_idl_db_get_mode(db, column) = 0;
+    *ovsdb_idl_get_mode(idl, column) &= ~(OVSDB_IDL_ALERT | OVSDB_IDL_TRACK);
 }
 
 /* Sets the mode for 'column' in 'idl' to 0.  See the big comment above
@@ -1805,7 +1075,7 @@ ovsdb_idl_db_omit(struct ovsdb_idl_db *db,
 void
 ovsdb_idl_omit(struct ovsdb_idl *idl, const struct ovsdb_idl_column *column)
 {
-    ovsdb_idl_db_omit(&idl->data, column);
+    *ovsdb_idl_get_mode(idl, column) = 0;
 }
 
 /* Returns the most recent IDL change sequence number that caused a
@@ -1816,7 +1086,7 @@ ovsdb_idl_table_get_seqno(const struct ovsdb_idl *idl,
                           const struct ovsdb_idl_table_class *table_class)
 {
     struct ovsdb_idl_table *table
-        = ovsdb_idl_db_table_from_class(&idl->data, table_class);
+        = ovsdb_idl_table_from_class(idl, table_class);
     unsigned int max_seqno = table->change_seqno[OVSDB_IDL_CHANGE_INSERT];
 
     if (max_seqno < table->change_seqno[OVSDB_IDL_CHANGE_MODIFY]) {
@@ -1852,461 +1122,156 @@ ovsdb_idl_row_get_seqno(const struct ovsdb_idl_row *row,
 void
 ovsdb_idl_track_add_column(struct ovsdb_idl *idl,
                            const struct ovsdb_idl_column *column)
-{
-    if (!(*ovsdb_idl_db_get_mode(&idl->data, column) & OVSDB_IDL_ALERT)) {
-        ovsdb_idl_add_column(idl, column);
-    }
-    *ovsdb_idl_db_get_mode(&idl->data, column) |= OVSDB_IDL_TRACK;
-}
-
-void
-ovsdb_idl_track_add_all(struct ovsdb_idl *idl)
-{
-    size_t i, j;
-
-    for (i = 0; i < idl->data.class_->n_tables; i++) {
-        const struct ovsdb_idl_table_class *tc = &idl->data.class_->tables[i];
-
-        for (j = 0; j < tc->n_columns; j++) {
-            const struct ovsdb_idl_column *column = &tc->columns[j];
-            ovsdb_idl_track_add_column(idl, column);
-        }
-    }
-}
-
-/* Returns true if 'table' has any tracked column. */
-static bool
-ovsdb_idl_track_is_set(struct ovsdb_idl_table *table)
-{
-    size_t i;
-
-    for (i = 0; i < table->class_->n_columns; i++) {
-        if (table->modes[i] & OVSDB_IDL_TRACK) {
-            return true;
-        }
-    }
-   return false;
-}
-
-/* Returns the first tracked row in table with class 'table_class'
- * for the specified 'idl'. Returns NULL if there are no tracked rows */
-const struct ovsdb_idl_row *
-ovsdb_idl_track_get_first(const struct ovsdb_idl *idl,
-                          const struct ovsdb_idl_table_class *table_class)
-{
-    struct ovsdb_idl_table *table
-        = ovsdb_idl_db_table_from_class(&idl->data, table_class);
-
-    if (!ovs_list_is_empty(&table->track_list)) {
-        return CONTAINER_OF(ovs_list_front(&table->track_list), struct ovsdb_idl_row, track_node);
-    }
-    return NULL;
-}
-
-/* Returns the next tracked row in table after the specified 'row'
- * (in no particular order). Returns NULL if there are no tracked rows */
-const struct ovsdb_idl_row *
-ovsdb_idl_track_get_next(const struct ovsdb_idl_row *row)
-{
-    if (row->track_node.next != &row->table->track_list) {
-        return CONTAINER_OF(row->track_node.next, struct ovsdb_idl_row, track_node);
-    }
-
-    return NULL;
-}
-
-/* Returns true if a tracked 'column' in 'row' was updated by IDL, false
- * otherwise. The tracking data is cleared by ovsdb_idl_track_clear()
- *
- * Function returns false if 'column' is not tracked (see
- * ovsdb_idl_track_add_column()).
- */
-bool
-ovsdb_idl_track_is_updated(const struct ovsdb_idl_row *row,
-                           const struct ovsdb_idl_column *column)
-{
-    const struct ovsdb_idl_table_class *class;
-    size_t column_idx;
-
-    class = row->table->class_;
-    column_idx = column - class->columns;
-
-    if (row->updated && bitmap_is_set(row->updated, column_idx)) {
-        return true;
-    } else {
-        return false;
-    }
-}
-
-/* Flushes the tracked rows. Client calls this function after calling
- * ovsdb_idl_run() and read all tracked rows with the ovsdb_idl_track_get_*()
- * functions. This is usually done at the end of the client's processing
- * loop when it is ready to do ovsdb_idl_run() again.
- */
-static void
-ovsdb_idl_db_track_clear(struct ovsdb_idl_db *db)
-{
-    size_t i;
-
-    for (i = 0; i < db->class_->n_tables; i++) {
-        struct ovsdb_idl_table *table = &db->tables[i];
-
-        if (!ovs_list_is_empty(&table->track_list)) {
-            struct ovsdb_idl_row *row, *next;
-
-            LIST_FOR_EACH_SAFE(row, next, track_node, &table->track_list) {
-                if (row->updated) {
-                    free(row->updated);
-                    row->updated = NULL;
-                }
-
-                row->change_seqno[OVSDB_IDL_CHANGE_INSERT] =
-                    row->change_seqno[OVSDB_IDL_CHANGE_MODIFY] =
-                    row->change_seqno[OVSDB_IDL_CHANGE_DELETE] = 0;
-
-                ovs_list_remove(&row->track_node);
-                ovs_list_init(&row->track_node);
-                if (ovsdb_idl_row_is_orphan(row) && row->tracked_old_datum) {
-                    ovsdb_idl_row_unparse(row);
-                    const struct ovsdb_idl_table_class *class =
-                                                        row->table->class_;
-                    for (size_t c = 0; c < class->n_columns; c++) {
-                        ovsdb_datum_destroy(&row->tracked_old_datum[c],
-                                            &class->columns[c].type);
-                    }
-                    free(row->tracked_old_datum);
-                    row->tracked_old_datum = NULL;
-                    free(row);
-                }
-            }
-        }
-    }
-}
-
-/* Flushes the tracked rows. Client calls this function after calling
- * ovsdb_idl_run() and read all tracked rows with the ovsdb_idl_track_get_*()
- * functions. This is usually done at the end of the client's processing
- * loop when it is ready to do ovsdb_idl_run() again.
- */
-void
-ovsdb_idl_track_clear(struct ovsdb_idl *idl)
-{
-    ovsdb_idl_db_track_clear(&idl->data);
-}
-
-static void
-ovsdb_idl_send_schema_request(struct ovsdb_idl *idl,
-                              struct ovsdb_idl_db *db)
-{
-    ovsdb_idl_send_request(idl, jsonrpc_create_request(
-                               "get_schema",
-                               json_array_create_1(json_string_create(
-                                                       db->class_->database)),
-                               NULL));
-}
-
-static void
-ovsdb_idl_send_db_change_aware(struct ovsdb_idl *idl)
-{
-    struct jsonrpc_msg *msg = jsonrpc_create_request(
-        "set_db_change_aware", json_array_create_1(json_boolean_create(true)),
-        NULL);
-    jsonrpc_session_send(idl->session, msg);
-}
-
-static bool
-ovsdb_idl_check_server_db(struct ovsdb_idl *idl)
-{
-    const struct serverrec_database *database;
-    SERVERREC_DATABASE_FOR_EACH (database, idl) {
-        if (uuid_is_zero(&idl->cid)
-            ? !strcmp(database->name, idl->data.class_->database)
-            : database->n_cid && uuid_equals(database->cid, &idl->cid)) {
-            break;
-        }
-    }
-
-    static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
-    const char *server_name = jsonrpc_session_get_name(idl->session);
-    bool ok = false;
-    if (!database) {
-        VLOG_INFO_RL(&rl, "%s: server does not have %s database",
-                     server_name, idl->data.class_->database);
-    } else if (!strcmp(database->model, "clustered")) {
-        uint64_t index = database->n_index ? *database->index : 0;
-
-        if (!database->schema) {
-            VLOG_INFO("%s: clustered database server has not yet joined "
-                      "cluster; trying another server", server_name);
-        } else if (!database->connected) {
-            VLOG_INFO("%s: clustered database server is disconnected "
-                      "from cluster; trying another server", server_name);
-        } else if (idl->leader_only && !database->leader) {
-            VLOG_INFO("%s: clustered database server is not cluster "
-                      "leader; trying another server", server_name);
-        } else if (index < idl->min_index) {
-            VLOG_WARN("%s: clustered database server has stale data; "
-                      "trying another server", server_name);
-        } else {
-            idl->min_index = index;
-            ok = true;
-        }
-    } else {
-        ok = true;
-    }
-    if (!ok) {
-        ovsdb_idl_retry(idl);
-        return false;
-    }
-
-    if (idl->state == IDL_S_SERVER_MONITOR_COND_REQUESTED) {
-        json_destroy(idl->data.schema);
-        idl->data.schema = json_from_string(database->schema);
-        ovsdb_idl_send_monitor_request(idl, &idl->data,
-                                       OVSDB_IDL_MM_MONITOR_COND_SINCE);
-        ovsdb_idl_transition(idl, IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED);
-    }
-    return true;
-}
-
-static void
-ovsdb_idl_send_monitor_request(struct ovsdb_idl *idl, struct ovsdb_idl_db *db,
-                               enum ovsdb_idl_monitor_method monitor_method)
-{
-    struct shash *schema = ovsdb_cs_parse_schema(db->schema);
-    struct json *monitor_requests = json_object_create();
-
-    for (size_t i = 0; i < db->class_->n_tables; i++) {
-        struct ovsdb_idl_table *table = &db->tables[i];
-        const struct ovsdb_idl_table_class *tc = table->class_;
-        struct json *monitor_request;
-        const struct sset *table_schema
-            = schema ? shash_find_data(schema, table->class_->name) : NULL;
-
-        struct json *columns
-            = table->need_table ? json_array_create_empty() : NULL;
-        for (size_t j = 0; j < tc->n_columns; j++) {
-            const struct ovsdb_idl_column *column = &tc->columns[j];
-            bool db_has_column = (table_schema &&
-                                  sset_contains(table_schema, column->name));
-            if (column->is_synthetic) {
-                if (db_has_column) {
-                    VLOG_WARN("%s table in %s database has synthetic "
-                              "column %s", table->class_->name,
-                              db->class_->database, column->name);
-                }
-            } else if (table->modes[j] & OVSDB_IDL_MONITOR) {
-                if (table_schema && !db_has_column) {
-                    VLOG_WARN("%s table in %s database lacks %s column "
-                              "(database needs upgrade?)",
-                              table->class_->name, db->class_->database,
-                              column->name);
-                    continue;
-                }
-                if (!columns) {
-                    columns = json_array_create_empty();
-                }
-                json_array_add(columns, json_string_create(column->name));
-            }
-        }
-
-        if (columns) {
-            if (schema && !table_schema) {
-                VLOG_WARN("%s database lacks %s table "
-                          "(database needs upgrade?)",
-                          db->class_->database, table->class_->name);
-                json_destroy(columns);
-                continue;
-            }
-
-            monitor_request = json_object_create();
-            json_object_put(monitor_request, "columns", columns);
-
-            /* Always use acked conditions when requesting
-             * monitor_cond/monitor_cond_since.
-             */
-            const struct ovsdb_idl_condition *cond = table->ack_cond;
-            if ((monitor_method == OVSDB_IDL_MM_MONITOR_COND ||
-                 monitor_method == OVSDB_IDL_MM_MONITOR_COND_SINCE) &&
-                cond && !ovsdb_idl_condition_is_true(cond)) {
-                json_object_put(monitor_request, "where",
-                                ovsdb_idl_condition_to_json(cond));
-            }
-            json_object_put(monitor_requests, tc->name,
-                            json_array_create_1(monitor_request));
-        }
-    }
-    ovsdb_cs_free_schema(schema);
-
-    struct json *params = json_array_create_3(
-                              json_string_create(db->class_->database),
-                              json_clone(db->monitor_id),
-                              monitor_requests);
-    const char *method;
-    switch (monitor_method) {
-        case OVSDB_IDL_MM_MONITOR:
-            method = "monitor";
-            break;
-        case OVSDB_IDL_MM_MONITOR_COND:
-            method = "monitor_cond";
-            break;
-        case OVSDB_IDL_MM_MONITOR_COND_SINCE:
-            method = "monitor_cond_since";
-            struct json *json_last_id = json_string_create_nocopy(
-                    xasprintf(UUID_FMT, UUID_ARGS(&db->last_id)));
-            json_array_add(params, json_last_id);
-            break;
-        default:
-            OVS_NOT_REACHED();
-    }
-
-    ovsdb_idl_send_request(idl, jsonrpc_create_request(method, params, NULL));
-}
-
-static void
-log_parse_update_error(struct ovsdb_error *error)
-{
-    if (!VLOG_DROP_WARN(&syntax_rl)) {
-        char *s = ovsdb_error_to_string(error);
-        VLOG_WARN_RL(&syntax_rl, "%s", s);
-        free(s);
+{
+    if (!(*ovsdb_idl_get_mode(idl, column) & OVSDB_IDL_ALERT)) {
+        ovsdb_idl_add_column(idl, column);
     }
-    ovsdb_error_destroy(error);
+    *ovsdb_idl_get_mode(idl, column) |= OVSDB_IDL_TRACK;
 }
 
-static void
-ovsdb_idl_db_parse_monitor_reply(struct ovsdb_idl_db *db,
-                                 const struct json *result, int version)
-{
-    db->change_seqno++;
-    const struct json *table_updates = result;
-    bool clear_db = true;
-    if (version == 3) {
-        if (result->type != JSON_ARRAY || result->array.n != 3) {
-            struct ovsdb_error *error = ovsdb_syntax_error(result, NULL,
-                                     "Response of monitor_cond_since must "
-                                     "be an array with 3 elements.");
-            log_parse_update_error(error);
-            return;
-        }
+void
+ovsdb_idl_track_add_all(struct ovsdb_idl *idl)
+{
+    size_t i, j;
 
-        bool found = json_boolean(result->array.elems[0]);
-        if (found) {
-            clear_db = false;
-        }
+    for (i = 0; i < idl->class_->n_tables; i++) {
+        const struct ovsdb_idl_table_class *tc = &idl->class_->tables[i];
 
-        const char *last_id = json_string(result->array.elems[1]);
-        if (!uuid_from_string(&db->last_id, last_id)) {
-            struct ovsdb_error *error = ovsdb_syntax_error(result, NULL,
-                                     "Last-id %s is not in UUID format.",
-                                     last_id);
-            log_parse_update_error(error);
-            return;
+        for (j = 0; j < tc->n_columns; j++) {
+            const struct ovsdb_idl_column *column = &tc->columns[j];
+            ovsdb_idl_track_add_column(idl, column);
         }
-
-        table_updates = result->array.elems[2];
     }
-    if (clear_db) {
-        ovsdb_idl_db_clear(db);
-    }
-    ovsdb_idl_db_parse_update(db, table_updates, version);
 }
 
+/* Returns true if 'table' has any tracked column. */
 static bool
-ovsdb_idl_db_parse_update_rpc(struct ovsdb_idl_db *db,
-                              const struct jsonrpc_msg *msg)
+ovsdb_idl_track_is_set(struct ovsdb_idl_table *table)
 {
-    if (msg->type != JSONRPC_NOTIFY) {
-        return false;
-    }
+    size_t i;
 
-    int version;
-    uint8_t n;
-    if (!strcmp(msg->method, "update")) {
-        version = 1;
-        n = 2;
-    } else if (!strcmp(msg->method, "update2")) {
-        version = 2;
-        n = 2;
-    } else if (!strcmp(msg->method, "update3")) {
-        version = 3;
-        n = 3;
-    } else {
-        return false;
+    for (i = 0; i < table->class_->n_columns; i++) {
+        if (table->modes[i] & OVSDB_IDL_TRACK) {
+            return true;
+        }
     }
+   return false;
+}
 
-    struct json *params = msg->params;
-    if (params->type != JSON_ARRAY || params->array.n != n) {
-        struct ovsdb_error *error = ovsdb_syntax_error(params, NULL,
-                                 "%s must be an array with %u elements.",
-                                 msg->method, n);
-        log_parse_update_error(error);
-        return false;
-    }
+/* Returns the first tracked row in table with class 'table_class'
+ * for the specified 'idl'. Returns NULL if there are no tracked rows */
+const struct ovsdb_idl_row *
+ovsdb_idl_track_get_first(const struct ovsdb_idl *idl,
+                          const struct ovsdb_idl_table_class *table_class)
+{
+    struct ovsdb_idl_table *table
+        = ovsdb_idl_table_from_class(idl, table_class);
 
-    if (!json_equal(params->array.elems[0], db->monitor_id)) {
-        return false;
+    if (!ovs_list_is_empty(&table->track_list)) {
+        return CONTAINER_OF(ovs_list_front(&table->track_list), struct ovsdb_idl_row, track_node);
     }
+    return NULL;
+}
 
-    struct json *table_updates = params->array.elems[1];
-    if (!strcmp(msg->method, "update3")) {
-        table_updates = params->array.elems[2];
-        const char *last_id = json_string(params->array.elems[1]);
-        if (!uuid_from_string(&db->last_id, last_id)) {
-            struct ovsdb_error *error = ovsdb_syntax_error(params, NULL,
-                                     "Last-id %s is not in UUID format.",
-                                     last_id);
-            log_parse_update_error(error);
-            return false;
-        }
+/* Returns the next tracked row in table after the specified 'row'
+ * (in no particular order). Returns NULL if there are no tracked rows */
+const struct ovsdb_idl_row *
+ovsdb_idl_track_get_next(const struct ovsdb_idl_row *row)
+{
+    if (row->track_node.next != &row->table->track_list) {
+        return CONTAINER_OF(row->track_node.next, struct ovsdb_idl_row, track_node);
     }
-    ovsdb_idl_db_parse_update(db, table_updates, version);
-    return true;
+
+    return NULL;
 }
 
-static bool
-ovsdb_idl_handle_monitor_canceled(struct ovsdb_idl *idl,
-                                  struct ovsdb_idl_db *db,
-                                  const struct jsonrpc_msg *msg)
-{
-    if (msg->type != JSONRPC_NOTIFY
-        || strcmp(msg->method, "monitor_canceled")
-        || msg->params->type != JSON_ARRAY
-        || msg->params->array.n != 1
-        || !json_equal(msg->params->array.elems[0], db->monitor_id)) {
+/* Returns true if a tracked 'column' in 'row' was updated by IDL, false
+ * otherwise. The tracking data is cleared by ovsdb_idl_track_clear()
+ *
+ * Function returns false if 'column' is not tracked (see
+ * ovsdb_idl_track_add_column()).
+ */
+bool
+ovsdb_idl_track_is_updated(const struct ovsdb_idl_row *row,
+                           const struct ovsdb_idl_column *column)
+{
+    const struct ovsdb_idl_table_class *class;
+    size_t column_idx;
+
+    class = row->table->class_;
+    column_idx = column - class->columns;
+
+    if (row->updated && bitmap_is_set(row->updated, column_idx)) {
+        return true;
+    } else {
         return false;
     }
+}
+
+/* Flushes the tracked rows. Client calls this function after calling
+ * ovsdb_idl_run() and read all tracked rows with the ovsdb_idl_track_get_*()
+ * functions. This is usually done at the end of the client's processing
+ * loop when it is ready to do ovsdb_idl_run() again.
+ */
+void
+ovsdb_idl_track_clear(struct ovsdb_idl *idl)
+{
+    size_t i;
 
-    db->monitoring = OVSDB_IDL_NOT_MONITORING;
+    for (i = 0; i < idl->class_->n_tables; i++) {
+        struct ovsdb_idl_table *table = &idl->tables[i];
 
-    /* Cancel the other monitor and restart the FSM from the top.
-     *
-     * Maybe a more sophisticated response would be better in some cases, but
-     * it doesn't seem worth optimizing yet.  (Although this is already more
-     * sophisticated than just dropping the connection and reconnecting.) */
-    struct ovsdb_idl_db *other_db
-        = db == &idl->data ? &idl->server : &idl->data;
-    if (other_db->monitoring) {
-        jsonrpc_session_send(
-            idl->session,
-            jsonrpc_create_request(
-                "monitor_cancel",
-                json_array_create_1(json_clone(other_db->monitor_id)), NULL));
-        other_db->monitoring = OVSDB_IDL_NOT_MONITORING;
-    }
-    ovsdb_idl_restart_fsm(idl);
+        if (!ovs_list_is_empty(&table->track_list)) {
+            struct ovsdb_idl_row *row, *next;
 
-    return true;
+            LIST_FOR_EACH_SAFE(row, next, track_node, &table->track_list) {
+                if (row->updated) {
+                    free(row->updated);
+                    row->updated = NULL;
+                }
+
+                row->change_seqno[OVSDB_IDL_CHANGE_INSERT] =
+                    row->change_seqno[OVSDB_IDL_CHANGE_MODIFY] =
+                    row->change_seqno[OVSDB_IDL_CHANGE_DELETE] = 0;
+
+                ovs_list_remove(&row->track_node);
+                ovs_list_init(&row->track_node);
+                if (ovsdb_idl_row_is_orphan(row) && row->tracked_old_datum) {
+                    ovsdb_idl_row_unparse(row);
+                    const struct ovsdb_idl_table_class *class =
+                                                        row->table->class_;
+                    for (size_t c = 0; c < class->n_columns; c++) {
+                        ovsdb_datum_destroy(&row->tracked_old_datum[c],
+                                            &class->columns[c].type);
+                    }
+                    free(row->tracked_old_datum);
+                    row->tracked_old_datum = NULL;
+                    free(row);
+                }
+            }
+        }
+    }
+}
+
+static void
+log_parse_update_error(struct ovsdb_error *error)
+{
+    if (!VLOG_DROP_WARN(&syntax_rl)) {
+        char *s = ovsdb_error_to_string(error);
+        VLOG_WARN_RL(&syntax_rl, "%s", s);
+        free(s);
+    }
+    ovsdb_error_destroy(error);
 }
 
 static struct ovsdb_error *
-ovsdb_idl_db_parse_update__(struct ovsdb_idl_db *db,
-                            const struct ovsdb_cs_db_update *du)
+ovsdb_idl_parse_update__(struct ovsdb_idl *idl,
+                         const struct ovsdb_cs_db_update *du)
 {
     for (size_t i = 0; i < du->n; i++) {
         const struct ovsdb_cs_table_update *tu = &du->table_updates[i];
 
-        struct ovsdb_idl_table *table = shash_find_data(&db->table_by_name,
+        struct ovsdb_idl_table *table = shash_find_data(&idl->table_by_name,
                                                         tu->table_name);
         if (!table) {
             return ovsdb_syntax_error(
@@ -2317,13 +1282,12 @@ ovsdb_idl_db_parse_update__(struct ovsdb_idl_db *db,
             const struct ovsdb_cs_row_update *ru = &tu->row_updates[j];
             switch (ovsdb_idl_process_update(table, ru)) {
             case OVSDB_IDL_UPDATE_DB_CHANGED:
-                db->change_seqno++;
+                idl->change_seqno++;
                 break;
             case OVSDB_IDL_UPDATE_NO_CHANGES:
                 break;
             case OVSDB_IDL_UPDATE_INCONSISTENT:
-                memset(&db->last_id, 0, sizeof db->last_id);
-                ovsdb_idl_retry(db->idl);
+                ovsdb_cs_flag_inconsistency(idl->cs);
                 return ovsdb_error(NULL,
                                    "row update received for inconsistent "
                                    "IDL: reconnecting IDL and resync all "
@@ -2336,15 +1300,25 @@ ovsdb_idl_db_parse_update__(struct ovsdb_idl_db *db,
 }
 
 static void
-ovsdb_idl_db_parse_update(struct ovsdb_idl_db *db,
-                          const struct json *json_table_updates,
-                          int version)
+ovsdb_idl_parse_update(struct ovsdb_idl *idl,
+                       const struct ovsdb_cs_update_event *update)
 {
     struct ovsdb_cs_db_update *du;
     struct ovsdb_error *error = ovsdb_cs_parse_db_update(
-        json_table_updates, version, &du);
+        update->table_updates, update->version, &du);
     if (!error) {
-        error = ovsdb_idl_db_parse_update__(db, du);
+        /* If this is the first update notification we've received, then always
+         * indicate a change, even if nothing actually changes, because we
+         * should signal a change for an initially empty database. */
+        if (!idl->change_seqno) {
+            idl->change_seqno++;
+        }
+
+        if (update->clear) {
+            ovsdb_idl_clear(idl);
+        }
+
+        error = ovsdb_idl_parse_update__(idl, du);
     }
     ovsdb_cs_db_update_destroy(du);
     if (error) {
@@ -2457,7 +1431,7 @@ add_tracked_change_for_references(struct ovsdb_idl_row *row)
 
             ref->change_seqno[OVSDB_IDL_CHANGE_MODIFY]
                 = ref->table->change_seqno[OVSDB_IDL_CHANGE_MODIFY]
-                = ref->table->db->change_seqno + 1;
+                = ref->table->idl->change_seqno + 1;
 
             add_tracked_change_for_references(ref);
         }
@@ -2479,6 +1453,7 @@ ovsdb_idl_row_change(struct ovsdb_idl_row *row, const struct shash *values,
     const struct ovsdb_idl_table_class *class = table->class_;
     struct shash_node *node;
     bool changed = false;
+
     SHASH_FOR_EACH (node, values) {
         const char *column_name = node->name;
         const struct ovsdb_idl_column *column;
@@ -2520,7 +1495,7 @@ ovsdb_idl_row_change(struct ovsdb_idl_row *row, const struct shash *values,
                     changed = true;
                     row->change_seqno[change]
                         = row->table->change_seqno[change]
-                        = row->table->db->change_seqno + 1;
+                        = row->table->idl->change_seqno + 1;
 
                     if (table->modes[column_idx] & OVSDB_IDL_TRACK) {
                         if (ovs_list_is_empty(&row->track_node) &&
@@ -2686,20 +1661,25 @@ ovsdb_idl_index_generic_comparer(const void *a,
     }
 }
 
-static struct ovsdb_idl_index *
-ovsdb_idl_db_index_create(struct ovsdb_idl_db *db,
-                          const struct ovsdb_idl_index_column *columns,
-                          size_t n)
+/* Creates a new index for the given 'idl' and with the 'n' specified
+ * 'columns'.
+ *
+ * All indexes must be created before the first call to ovsdb_idl_run(). */
+struct ovsdb_idl_index *
+ovsdb_idl_index_create(struct ovsdb_idl *idl,
+                       const struct ovsdb_idl_index_column *columns,
+                       size_t n)
 {
     ovs_assert(n > 0);
 
     struct ovsdb_idl_index *index = xzalloc(sizeof *index);
 
-    index->table = ovsdb_idl_table_from_column(db, columns[0].column);
+    index->table = ovsdb_idl_table_from_column(idl, columns[0].column);
     for (size_t i = 0; i < n; i++) {
         const struct ovsdb_idl_index_column *c = &columns[i];
-        ovs_assert(ovsdb_idl_table_from_column(db, c->column) == index->table);
-        ovs_assert(*ovsdb_idl_db_get_mode(db, c->column) & OVSDB_IDL_MONITOR);
+        ovs_assert(ovsdb_idl_table_from_column(idl,
+                                               c->column) == index->table);
+        ovs_assert(*ovsdb_idl_get_mode(idl, c->column) & OVSDB_IDL_MONITOR);
     }
 
     index->columns = xmemdup(columns, n * sizeof *columns);
@@ -2711,18 +1691,6 @@ ovsdb_idl_db_index_create(struct ovsdb_idl_db *db,
     return index;
 }
 
-/* Creates a new index for the given 'idl' and with the 'n' specified
- * 'columns'.
- *
- * All indexes must be created before the first call to ovsdb_idl_run(). */
-struct ovsdb_idl_index *
-ovsdb_idl_index_create(struct ovsdb_idl *idl,
-                       const struct ovsdb_idl_index_column *columns,
-                       size_t n)
-{
-    return ovsdb_idl_db_index_create(&idl->data, columns, n);
-}
-
 struct ovsdb_idl_index *
 ovsdb_idl_index_create1(struct ovsdb_idl *idl,
                         const struct ovsdb_idl_column *column1)
@@ -3067,7 +2035,7 @@ ovsdb_idl_row_destroy(struct ovsdb_idl_row *row)
         if (ovsdb_idl_track_is_set(row->table)) {
             row->change_seqno[OVSDB_IDL_CHANGE_DELETE]
                 = row->table->change_seqno[OVSDB_IDL_CHANGE_DELETE]
-                = row->table->db->change_seqno + 1;
+                = row->table->idl->change_seqno + 1;
         }
         if (ovs_list_is_empty(&row->track_node)) {
             ovs_list_push_back(&row->table->track_list, &row->track_node);
@@ -3118,12 +2086,10 @@ ovsdb_idl_destroy_all_set_op_lists(struct ovsdb_idl_row *row)
 }
 
 static void
-ovsdb_idl_row_destroy_postprocess(struct ovsdb_idl_db *db)
+ovsdb_idl_row_destroy_postprocess(struct ovsdb_idl *idl)
 {
-    size_t i;
-
-    for (i = 0; i < db->class_->n_tables; i++) {
-        struct ovsdb_idl_table *table = &db->tables[i];
+    for (size_t i = 0; i < idl->class_->n_tables; i++) {
+        struct ovsdb_idl_table *table = &idl->tables[i];
 
         if (!ovs_list_is_empty(&table->track_list)) {
             struct ovsdb_idl_row *row, *next;
@@ -3211,26 +2177,12 @@ may_add_arc(const struct ovsdb_idl_row *src, const struct ovsdb_idl_row *dst)
     return arc->src != src;
 }
 
-static struct ovsdb_idl_table *
-ovsdb_idl_db_table_from_class(const struct ovsdb_idl_db *db,
-                              const struct ovsdb_idl_table_class *table_class)
-{
-    ptrdiff_t idx = table_class - db->class_->tables;
-    return idx >= 0 && idx < db->class_->n_tables ? &db->tables[idx] : NULL;
-}
-
 static struct ovsdb_idl_table *
 ovsdb_idl_table_from_class(const struct ovsdb_idl *idl,
                            const struct ovsdb_idl_table_class *table_class)
 {
-    struct ovsdb_idl_table *table;
-
-    table = ovsdb_idl_db_table_from_class(&idl->data, table_class);
-    if (!table) {
-         table = ovsdb_idl_db_table_from_class(&idl->server, table_class);
-    }
-
-    return table;
+    ptrdiff_t idx = table_class - idl->class_->tables;
+    return idx >= 0 && idx < idl->class_->n_tables ? &idl->tables[idx] : NULL;
 }
 
 /* Called by ovsdb-idlc generated code. */
@@ -3239,14 +2191,14 @@ ovsdb_idl_get_row_arc(struct ovsdb_idl_row *src,
                       const struct ovsdb_idl_table_class *dst_table_class,
                       const struct uuid *dst_uuid)
 {
-    struct ovsdb_idl_db *db = src->table->db;
+    struct ovsdb_idl *idl = src->table->idl;
     struct ovsdb_idl_table *dst_table;
     struct ovsdb_idl_arc *arc;
     struct ovsdb_idl_row *dst;
 
-    dst_table = ovsdb_idl_db_table_from_class(db, dst_table_class);
+    dst_table = ovsdb_idl_table_from_class(idl, dst_table_class);
     dst = ovsdb_idl_get_row(dst_table, dst_uuid);
-    if (db->txn || is_index_row(src)) {
+    if (idl->txn || is_index_row(src)) {
         /* There are two cases we should not update any arcs:
          *
          * 1. We're being called from ovsdb_idl_txn_write(). We must not update
@@ -3451,10 +2403,10 @@ ovsdb_idl_txn_create(struct ovsdb_idl *idl)
 {
     struct ovsdb_idl_txn *txn;
 
-    ovs_assert(!idl->data.txn);
-    idl->data.txn = txn = xmalloc(sizeof *txn);
+    ovs_assert(!idl->txn);
+    idl->txn = txn = xmalloc(sizeof *txn);
     txn->request_id = NULL;
-    txn->db = &idl->data;
+    txn->idl = idl;
     hmap_init(&txn->txn_rows);
     txn->status = TXN_UNCOMMITTED;
     txn->error = NULL;
@@ -3544,10 +2496,11 @@ ovsdb_idl_txn_destroy(struct ovsdb_idl_txn *txn)
 {
     struct ovsdb_idl_txn_insert *insert, *next;
 
-    json_destroy(txn->request_id);
     if (txn->status == TXN_INCOMPLETE) {
-        hmap_remove(&txn->db->outstanding_txns, &txn->hmap_node);
+        ovsdb_cs_forget_transaction(txn->idl->cs, txn->request_id);
+        hmap_remove(&txn->idl->outstanding_txns, &txn->hmap_node);
     }
+    json_destroy(txn->request_id);
     ovsdb_idl_txn_abort(txn);
     ds_destroy(&txn->comment);
     free(txn->error);
@@ -3642,7 +2595,7 @@ ovsdb_idl_txn_disassemble(struct ovsdb_idl_txn *txn)
      * ovsdb_idl_column's 'parse' function, which will call
      * ovsdb_idl_get_row_arc(), which will seen that the IDL is in a
      * transaction and fail to update the graph.  */
-    txn->db->txn = NULL;
+    txn->idl->txn = NULL;
 
     HMAP_FOR_EACH_SAFE (row, next, txn_node, &txn->txn_rows) {
         enum { INSERTED, MODIFIED, DELETED } op
@@ -3935,39 +2888,22 @@ ovsdb_idl_txn_extract_mutations(struct ovsdb_idl_row *row,
 enum ovsdb_idl_txn_status
 ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
 {
-    struct ovsdb_idl_row *row;
-    struct json *operations;
-    bool any_updates;
-
-    if (txn != txn->db->txn) {
+    struct ovsdb_idl *idl = txn->idl;
+    if (txn != idl->txn) {
         goto coverage_out;
-    }
-
-    /* If we're still connecting or re-connecting, don't bother sending a
-     * transaction. */
-    if (txn->db->idl->state != IDL_S_MONITORING) {
+    } else if (!ovsdb_cs_may_send_transaction(idl->cs)) {
         txn->status = TXN_TRY_AGAIN;
         goto disassemble_out;
-    }
-
-    /* If we need a lock but don't have it, give up quickly. */
-    if (txn->db->lock_name && !txn->db->has_lock) {
+    } else if (ovsdb_cs_get_lock(idl->cs) && !ovsdb_cs_has_lock(idl->cs)) {
         txn->status = TXN_NOT_LOCKED;
         goto disassemble_out;
     }
 
-    operations = json_array_create_1(
-        json_string_create(txn->db->class_->database));
-
-    /* Assert that we have the required lock (avoiding a race). */
-    if (txn->db->lock_name) {
-        struct json *op = json_object_create();
-        json_array_add(operations, op);
-        json_object_put_string(op, "op", "assert");
-        json_object_put_string(op, "lock", txn->db->lock_name);
-    }
+    struct json *operations = json_array_create_1(
+        json_string_create(idl->class_->database));
 
     /* Add prerequisites and declarations of new rows. */
+    struct ovsdb_idl_row *row;
     HMAP_FOR_EACH (row, txn_node, &txn->txn_rows) {
         /* XXX check that deleted rows exist even if no prereqs? */
         if (row->prereqs) {
@@ -3999,15 +2935,15 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
     }
 
     /* Add updates. */
-    any_updates = false;
+    bool any_updates = false;
 
     /* For tables constrained to have only a single row (a fairly common OVSDB
      * pattern for storing global data), identify whether we're inserting a
      * row.  If so, then verify that the table is empty before inserting the
      * row.  This gives us a clear verification-related failure if there was an
      * insertion race with another client. */
-    for (size_t i = 0; i < txn->db->class_->n_tables; i++) {
-        struct ovsdb_idl_table *table = &txn->db->tables[i];
+    for (size_t i = 0; i < idl->class_->n_tables; i++) {
+        struct ovsdb_idl_table *table = &idl->tables[i];
         if (table->class_->is_singleton) {
             /* Count the number of rows in the table before and after our
              * transaction commits.  This is O(n) in the number of rows in the
@@ -4182,18 +3118,15 @@ ovsdb_idl_txn_commit(struct ovsdb_idl_txn *txn)
     if (!any_updates) {
         txn->status = TXN_UNCHANGED;
         json_destroy(operations);
-    } else if (!txn->db->idl->session) {
-        txn->status = TXN_TRY_AGAIN;
-        json_destroy(operations);
-    } else if (!jsonrpc_session_send(
-                   txn->db->idl->session,
-                   jsonrpc_create_request(
-                       "transact", operations, &txn->request_id))) {
-        hmap_insert(&txn->db->outstanding_txns, &txn->hmap_node,
-                    json_hash(txn->request_id, 0));
-        txn->status = TXN_INCOMPLETE;
     } else {
-        txn->status = TXN_TRY_AGAIN;
+        txn->request_id = ovsdb_cs_send_transaction(idl->cs, operations);
+        if (txn->request_id) {
+            hmap_insert(&idl->outstanding_txns, &txn->hmap_node,
+                        json_hash(txn->request_id, 0));
+            txn->status = TXN_INCOMPLETE;
+        } else {
+            txn->status = TXN_TRY_AGAIN;
+        }
     }
 
 disassemble_out:
@@ -4226,8 +3159,8 @@ ovsdb_idl_txn_commit_block(struct ovsdb_idl_txn *txn)
 
     fatal_signal_run();
     while ((status = ovsdb_idl_txn_commit(txn)) == TXN_INCOMPLETE) {
-        ovsdb_idl_run(txn->db->idl);
-        ovsdb_idl_wait(txn->db->idl);
+        ovsdb_idl_run(txn->idl);
+        ovsdb_idl_wait(txn->idl);
         ovsdb_idl_txn_wait(txn);
         poll_block();
     }
@@ -4318,7 +3251,7 @@ ovsdb_idl_txn_complete(struct ovsdb_idl_txn *txn,
                        enum ovsdb_idl_txn_status status)
 {
     txn->status = status;
-    hmap_remove(&txn->db->outstanding_txns, &txn->hmap_node);
+    hmap_remove(&txn->idl->outstanding_txns, &txn->hmap_node);
 }
 
 static void
@@ -4345,7 +3278,7 @@ ovsdb_idl_txn_write__(const struct ovsdb_idl_row *row_,
     ovs_assert(row->old_datum == NULL ||
                row->table->modes[column_idx] & OVSDB_IDL_MONITOR);
 
-    if (row->table->db->verify_write_only && !write_only) {
+    if (row->table->idl->verify_write_only && !write_only) {
         VLOG_ERR("Bug: Attempt to write to a read/write column (%s:%s) when"
                  " explicitly configured not to.", class->name, column->name);
         goto discard_datum;
@@ -4372,7 +3305,7 @@ ovsdb_idl_txn_write__(const struct ovsdb_idl_row *row_,
         ovsdb_idl_remove_from_indexes(row);
     }
     if (hmap_node_is_null(&row->txn_node)) {
-        hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+        hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
                     uuid_hash(&row->uuid));
     }
     if (row->old_datum == row->new_datum) {
@@ -4498,7 +3431,7 @@ ovsdb_idl_txn_verify(const struct ovsdb_idl_row *row_,
     }
 
     if (hmap_node_is_null(&row->txn_node)) {
-        hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+        hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
                     uuid_hash(&row->uuid));
     }
     if (!row->prereqs) {
@@ -4531,12 +3464,12 @@ ovsdb_idl_txn_delete(const struct ovsdb_idl_row *row_)
         ovsdb_idl_row_clear_new(row);
         ovs_assert(!row->prereqs);
         hmap_remove(&row->table->rows, &row->hmap_node);
-        hmap_remove(&row->table->db->txn->txn_rows, &row->txn_node);
+        hmap_remove(&row->table->idl->txn->txn_rows, &row->txn_node);
         free(row);
         return;
     }
     if (hmap_node_is_null(&row->txn_node)) {
-        hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+        hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
                     uuid_hash(&row->uuid));
     }
     ovsdb_idl_row_clear_new(row);
@@ -4569,7 +3502,7 @@ ovsdb_idl_txn_insert(struct ovsdb_idl_txn *txn,
         uuid_generate(&row->uuid);
     }
 
-    row->table = ovsdb_idl_db_table_from_class(txn->db, class);
+    row->table = ovsdb_idl_table_from_class(txn->idl, class);
     row->new_datum = xmalloc(class->n_columns * sizeof *row->new_datum);
     hmap_insert(&row->table->rows, &row->hmap_node, uuid_hash(&row->uuid));
     hmap_insert(&txn->txn_rows, &row->txn_node, uuid_hash(&row->uuid));
@@ -4579,29 +3512,22 @@ ovsdb_idl_txn_insert(struct ovsdb_idl_txn *txn,
 }
 
 static void
-ovsdb_idl_db_txn_abort_all(struct ovsdb_idl_db *db)
+ovsdb_idl_txn_abort_all(struct ovsdb_idl *idl)
 {
     struct ovsdb_idl_txn *txn;
 
-    HMAP_FOR_EACH (txn, hmap_node, &db->outstanding_txns) {
+    HMAP_FOR_EACH (txn, hmap_node, &idl->outstanding_txns) {
         ovsdb_idl_txn_complete(txn, TXN_TRY_AGAIN);
     }
 }
 
-static void
-ovsdb_idl_txn_abort_all(struct ovsdb_idl *idl)
-{
-    ovsdb_idl_db_txn_abort_all(&idl->server);
-    ovsdb_idl_db_txn_abort_all(&idl->data);
-}
-
 static struct ovsdb_idl_txn *
-ovsdb_idl_db_txn_find(struct ovsdb_idl_db *db, const struct json *id)
+ovsdb_idl_txn_find(struct ovsdb_idl *idl, const struct json *id)
 {
     struct ovsdb_idl_txn *txn;
 
     HMAP_FOR_EACH_WITH_HASH (txn, hmap_node,
-                             json_hash(id, 0), &db->outstanding_txns) {
+                             json_hash(id, 0), &idl->outstanding_txns) {
         if (json_equal(id, txn->request_id)) {
             return txn;
         }
@@ -4640,7 +3566,7 @@ ovsdb_idl_txn_process_inc_reply(struct ovsdb_idl_txn *txn,
     }
 
     /* We know that this is a JSON object because the loop in
-     * ovsdb_idl_db_txn_process_reply() checked. */
+     * ovsdb_idl_txn_process_reply() checked. */
     mutate = json_object(results->elems[txn->inc_index]);
     count = shash_find_data(mutate, "count");
     if (!check_json_type(count, JSON_INTEGER, "\"mutate\" reply \"count\"")) {
@@ -4716,18 +3642,16 @@ ovsdb_idl_txn_process_insert_reply(struct ovsdb_idl_txn_insert *insert,
     return true;
 }
 
-static bool
-ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *db,
-                               const struct jsonrpc_msg *msg)
+static void
+ovsdb_idl_txn_process_reply(struct ovsdb_idl *idl,
+                            const struct jsonrpc_msg *msg)
 {
-    struct ovsdb_idl_txn *txn;
-    enum ovsdb_idl_txn_status status;
-
-    txn = ovsdb_idl_db_txn_find(db, msg->id);
+    struct ovsdb_idl_txn *txn = ovsdb_idl_txn_find(idl, msg->id);
     if (!txn) {
-        return false;
+        return;
     }
 
+    enum ovsdb_idl_txn_status status;
     if (msg->type == JSONRPC_ERROR) {
         if (msg->error
             && msg->error->type == JSON_STRING
@@ -4768,7 +3692,7 @@ ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *db,
                             soft_errors++;
                         } else if (!strcmp(error->string,
                                            "unknown database")) {
-                            ovsdb_idl_retry(db->idl);
+                            ovsdb_cs_flag_inconsistency(idl->cs);
                             soft_errors++;
                         } else if (!strcmp(error->string, "not owner")) {
                             lock_errors++;
@@ -4817,7 +3741,6 @@ ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *db,
     }
 
     ovsdb_idl_txn_complete(txn, status);
-    return true;
 }
 
 /* Returns the transaction currently active for 'row''s IDL.  A transaction
@@ -4825,7 +3748,7 @@ ovsdb_idl_db_txn_process_reply(struct ovsdb_idl_db *db,
 struct ovsdb_idl_txn *
 ovsdb_idl_txn_get(const struct ovsdb_idl_row *row)
 {
-    struct ovsdb_idl_txn *txn = row->table->db->txn;
+    struct ovsdb_idl_txn *txn = row->table->idl->txn;
     ovs_assert(txn != NULL);
     return txn;
 }
@@ -4834,7 +3757,7 @@ ovsdb_idl_txn_get(const struct ovsdb_idl_row *row)
 struct ovsdb_idl *
 ovsdb_idl_txn_get_idl (struct ovsdb_idl_txn *txn)
 {
-    return txn->db->idl;
+    return txn->idl;
 }
 
 /* Blocks until 'idl' successfully connects to the remote database and
@@ -4852,31 +3775,6 @@ ovsdb_idl_get_initial_snapshot(struct ovsdb_idl *idl)
     }
 }
 
-static struct jsonrpc_msg *
-ovsdb_idl_db_set_lock(struct ovsdb_idl_db *db, const char *lock_name)
-{
-    ovs_assert(!db->txn);
-    ovs_assert(hmap_is_empty(&db->outstanding_txns));
-
-    if (db->lock_name
-        && (!lock_name || strcmp(lock_name, db->lock_name))) {
-        /* Release previous lock. */
-        struct jsonrpc_msg *msg = ovsdb_idl_db_compose_unlock_request(db);
-        free(db->lock_name);
-        db->lock_name = NULL;
-        db->is_lock_contended = false;
-        return msg;
-    }
-
-    if (lock_name && !db->lock_name) {
-        /* Acquire new lock. */
-        db->lock_name = xstrdup(lock_name);
-        return ovsdb_idl_db_compose_lock_request(db);
-    }
-
-    return NULL;
-}
-
 /* If 'lock_name' is nonnull, configures 'idl' to obtain the named lock from
  * the database server and to avoid modifying the database when the lock cannot
  * be acquired (that is, when another client has the same lock).
@@ -4886,17 +3784,7 @@ ovsdb_idl_db_set_lock(struct ovsdb_idl_db *db, const char *lock_name)
 void
 ovsdb_idl_set_lock(struct ovsdb_idl *idl, const char *lock_name)
 {
-    for (;;) {
-        struct jsonrpc_msg *msg = ovsdb_idl_db_set_lock(&idl->data, lock_name);
-        if (!msg) {
-            break;
-        }
-        if (idl->session) {
-            jsonrpc_session_send(idl->session, msg);
-        } else {
-            jsonrpc_msg_destroy(msg);
-        }
-    }
+    ovsdb_cs_set_lock(idl->cs, lock_name);
 }
 
 /* Returns true if 'idl' is configured to obtain a lock and owns that lock.
@@ -4908,7 +3796,7 @@ ovsdb_idl_set_lock(struct ovsdb_idl *idl, const char *lock_name)
 bool
 ovsdb_idl_has_lock(const struct ovsdb_idl *idl)
 {
-    return idl->data.has_lock;
+    return ovsdb_cs_has_lock(idl->cs);
 }
 
 /* Returns true if 'idl' is configured to obtain a lock but the database server
@@ -4916,122 +3804,7 @@ ovsdb_idl_has_lock(const struct ovsdb_idl *idl)
 bool
 ovsdb_idl_is_lock_contended(const struct ovsdb_idl *idl)
 {
-    return idl->data.is_lock_contended;
-}
-
-static void
-ovsdb_idl_db_update_has_lock(struct ovsdb_idl_db *db, bool new_has_lock)
-{
-    if (new_has_lock && !db->has_lock) {
-        if (db->idl->state == IDL_S_MONITORING) {
-            db->change_seqno++;
-        } else {
-            /* We're setting up a session, so don't signal that the database
-             * changed.  Finalizing the session will increment change_seqno
-             * anyhow. */
-        }
-        db->is_lock_contended = false;
-    }
-    db->has_lock = new_has_lock;
-}
-
-static bool
-ovsdb_idl_db_process_lock_replies(struct ovsdb_idl_db *db,
-                                  const struct jsonrpc_msg *msg)
-{
-    if (msg->type == JSONRPC_REPLY
-        && db->lock_request_id
-        && json_equal(db->lock_request_id, msg->id)) {
-        /* Reply to our "lock" request. */
-        ovsdb_idl_db_parse_lock_reply(db, msg->result);
-        return true;
-    }
-
-    if (msg->type == JSONRPC_NOTIFY) {
-        if (!strcmp(msg->method, "locked")) {
-            /* We got our lock. */
-            return ovsdb_idl_db_parse_lock_notify(db, msg->params, true);
-        } else if (!strcmp(msg->method, "stolen")) {
-            /* Someone else stole our lock. */
-            return ovsdb_idl_db_parse_lock_notify(db, msg->params, false);
-        }
-    }
-
-    return false;
-}
-
-static struct jsonrpc_msg *
-ovsdb_idl_db_compose_lock_request__(struct ovsdb_idl_db *db,
-                                    const char *method)
-{
-    ovsdb_idl_db_update_has_lock(db, false);
-
-    json_destroy(db->lock_request_id);
-    db->lock_request_id = NULL;
-
-    struct json *params = json_array_create_1(json_string_create(
-                                                  db->lock_name));
-    return jsonrpc_create_request(method, params, NULL);
-}
-
-static struct jsonrpc_msg *
-ovsdb_idl_db_compose_lock_request(struct ovsdb_idl_db *db)
-{
-    struct jsonrpc_msg *msg = ovsdb_idl_db_compose_lock_request__(db, "lock");
-    db->lock_request_id = json_clone(msg->id);
-    return msg;
-}
-
-static struct jsonrpc_msg *
-ovsdb_idl_db_compose_unlock_request(struct ovsdb_idl_db *db)
-{
-    return ovsdb_idl_db_compose_lock_request__(db, "unlock");
-}
-
-static void
-ovsdb_idl_db_parse_lock_reply(struct ovsdb_idl_db *db,
-                              const struct json *result)
-{
-    bool got_lock;
-
-    json_destroy(db->lock_request_id);
-    db->lock_request_id = NULL;
-
-    if (result->type == JSON_OBJECT) {
-        const struct json *locked;
-
-        locked = shash_find_data(json_object(result), "locked");
-        got_lock = locked && locked->type == JSON_TRUE;
-    } else {
-        got_lock = false;
-    }
-
-    ovsdb_idl_db_update_has_lock(db, got_lock);
-    if (!got_lock) {
-        db->is_lock_contended = true;
-    }
-}
-
-static bool
-ovsdb_idl_db_parse_lock_notify(struct ovsdb_idl_db *db,
-                               const struct json *params,
-                               bool new_has_lock)
-{
-    if (db->lock_name
-        && params->type == JSON_ARRAY
-        && json_array(params)->n > 0
-        && json_array(params)->elems[0]->type == JSON_STRING) {
-        const char *lock_name = json_string(json_array(params)->elems[0]);
-
-        if (!strcmp(db->lock_name, lock_name)) {
-            ovsdb_idl_db_update_has_lock(db, new_has_lock);
-            if (!new_has_lock) {
-                db->is_lock_contended = true;
-            }
-            return true;
-        }
-    }
-    return false;
+    return ovsdb_cs_is_lock_contended(idl->cs);
 }
 
 /* Inserts a new Map Operation into current transaction. */
@@ -5065,7 +3838,7 @@ ovsdb_idl_txn_add_map_op(struct ovsdb_idl_row *row,
 
     /* Add this row to transaction's list of rows. */
     if (hmap_node_is_null(&row->txn_node)) {
-        hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+        hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
                     uuid_hash(&row->uuid));
     }
 }
@@ -5101,7 +3874,7 @@ ovsdb_idl_txn_add_set_op(struct ovsdb_idl_row *row,
 
     /* Add this row to the transactions's list of rows. */
     if (hmap_node_is_null(&row->txn_node)) {
-        hmap_insert(&row->table->db->txn->txn_rows, &row->txn_node,
+        hmap_insert(&row->table->idl->txn->txn_rows, &row->txn_node,
                     uuid_hash(&row->uuid));
     }
 }
