Re: Logical Replication WIP

/* * Replication slot on-disk data structure.
@@ -225,10 +226,25 @@ ReplicationSlotCreate(const char *name, bool db_specific,    ReplicationSlot *slot = NULL;    int
          i;
 

-    Assert(MyReplicationSlot == NULL);
+    /* Only aka ephemeral slots can survive across commands. */

What does this comment mean?


+    Assert(!MyReplicationSlot ||
+           MyReplicationSlot->data.persistency == RS_EPHEMERAL);

+    if (MyReplicationSlot)
+    {
+        /* Already acquired? Nothis to do. */

typo.

+        if (namestrcmp(&MyReplicationSlot->data.name, name) == 0)
+            return;
+
+        ereport(ERROR,
+                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+                 errmsg("cannot create replication slot %s, another slot %s is "
+                        "already active in this session",
+                        name, NameStr(MyReplicationSlot->data.name))));
+    }
+

Why do we now create slots that are already created? That seems like an
odd API change.
    /*     * If some other backend ran this code concurrently with us, we'd likely     * both allocate the same slot,
andthat would be bad.  We'd also be at
 
@@ -331,10 +347,25 @@ ReplicationSlotAcquire(const char *name)    int            i;    int            active_pid = 0;

-    Assert(MyReplicationSlot == NULL);
+    /* Only aka ephemeral slots can survive across commands. */
+    Assert(!MyReplicationSlot ||
+           MyReplicationSlot->data.persistency == RS_EPHEMERAL);
    ReplicationSlotValidateName(name, ERROR);

+    if (MyReplicationSlot)
+    {
+        /* Already acquired? Nothis to do. */
+        if (namestrcmp(&MyReplicationSlot->data.name, name) == 0)
+            return;
+
+        ereport(ERROR,
+                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+                 errmsg("cannot acquire replication slot %s, another slot %s is "
+                        "already active in this session",
+                        name, NameStr(MyReplicationSlot->data.name))));
+    }
+    /* Search for the named slot and mark it active if we find it. */    LWLockAcquire(ReplicationSlotControlLock,
LW_SHARED);   for (i = 0; i < max_replication_slots; i++)
 
@@ -406,12 +437,26 @@ ReplicationSlotRelease(void)}

Uh? We shouldn't ever have to acquire ephemeral
/*
+ * Same as above but only if currently acquired slot is peristent one.
+ */

s/peristent/persistent/

+void
+ReplicationSlotReleasePersistent(void)
+{
+    Assert(MyReplicationSlot);
+
+    if (MyReplicationSlot->data.persistency == RS_PERSISTENT)
+        ReplicationSlotRelease();
+}

Ick.



Hm. I think I have to agree a bit with Peter here.  Overloading
MyReplicationSlot this way seems ugly, and I think there's a bunch of
bugs around it too.


Sounds what we really want is a) two different lifetimes for ephemeral
slots, session and "command" b) have a number of slots that are released
either after a failed transaction / command or at session end.   The
easiest way for that appears to have a list of slots to be checked at
end-of-xact and backend shutdown. 

Regards,

Andres