diff --git a/include/qpid/dispatch/router_core.h b/include/qpid/dispatch/router_core.h
index 3d0e270..0d66d1c 100644
--- a/include/qpid/dispatch/router_core.h
+++ b/include/qpid/dispatch/router_core.h
@@ -553,7 +553,7 @@
 typedef void (*qdr_link_offer_t)         (void *context, qdr_link_t *link, int delivery_count);
 typedef void (*qdr_link_drained_t)       (void *context, qdr_link_t *link);
 typedef void (*qdr_link_drain_t)         (void *context, qdr_link_t *link, bool mode);
-typedef void (*qdr_link_push_t)          (void *context, qdr_link_t *link);
+typedef int  (*qdr_link_push_t)          (void *context, qdr_link_t *link, int limit);
 typedef void (*qdr_link_deliver_t)       (void *context, qdr_link_t *link, qdr_delivery_t *delivery, bool settled);
 typedef void (*qdr_delivery_update_t)    (void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled);
 
diff --git a/src/router_core/connections.c b/src/router_core/connections.c
index 8c95f9f..abc6a4c 100644
--- a/src/router_core/connections.c
+++ b/src/router_core/connections.c
@@ -201,22 +201,6 @@
 
     int event_count = 0;
 
-    do {
-        sys_mutex_lock(conn->work_lock);
-        ref = DEQ_HEAD(conn->links_with_deliveries);
-        if (ref) {
-            link = ref->link;
-            qdr_del_link_ref(&conn->links_with_deliveries, ref->link, QDR_LINK_LIST_CLASS_DELIVERY);
-        } else
-            link = 0;
-        sys_mutex_unlock(conn->work_lock);
-
-        if (link) {
-            core->push_handler(core->user_context, link);
-            event_count++;
-        }
-    } while (link);
-
     sys_mutex_lock(conn->work_lock);
     DEQ_MOVE(conn->work_list, work_list);
     sys_mutex_unlock(conn->work_lock);
@@ -234,55 +218,93 @@
         case QDR_CONNECTION_WORK_SECOND_ATTACH :
             core->second_attach_handler(core->user_context, work->link, work->source, work->target);
             break;
-
-        case QDR_CONNECTION_WORK_FIRST_DETACH :
-            core->detach_handler(core->user_context, work->link, work->error, true, work->close_link);
-            break;
-
-        case QDR_CONNECTION_WORK_SECOND_DETACH :
-            core->detach_handler(core->user_context, work->link, work->error, false, work->close_link);
-            free_qdr_link_t(work->link);
-            break;
         }
 
         qdr_terminus_free(work->source);
         qdr_terminus_free(work->target);
-        qdr_error_free(work->error);
         free_qdr_connection_work_t(work);
 
         work = DEQ_HEAD(work_list);
     }
 
     do {
+        qdr_link_work_t *link_work;
+
         sys_mutex_lock(conn->work_lock);
-        ref = DEQ_HEAD(conn->links_with_credit);
+        ref = DEQ_HEAD(conn->links_with_work);
         if (ref) {
             link = ref->link;
-            qdr_del_link_ref(&conn->links_with_credit, ref->link, QDR_LINK_LIST_CLASS_FLOW);
+            qdr_del_link_ref(&conn->links_with_work, ref->link, QDR_LINK_LIST_CLASS_WORK);
+            link_work = DEQ_HEAD(link->work_list);
+            if (link_work)
+                DEQ_REMOVE_HEAD(link->work_list);
         } else
             link = 0;
         sys_mutex_unlock(conn->work_lock);
 
         if (link) {
-            if (link->incremental_credit > 0) {
-                core->flow_handler(core->user_context, link, link->incremental_credit);
-                link->incremental_credit = 0;
+            while (link_work) {
+                switch (link_work->work_type) {
+                case QDR_LINK_WORK_DELIVERY :
+                    {
+                        int count = core->push_handler(core->user_context, link, link_work->value);
+                        assert(count <= link_work->value);
+                        link_work->value -= count;
+                        break;
+                    }
 
-                //
-                // Note:  This unprotected read of a CT-only value is safe in this case.
-                // If there is pending credit on the link that needs to be pushed down to
-                // Proton, we need to give the core a kick to make sure it is sent.  It is
-                // possible that no more credit will be issued to cause the movement of CT
-                // credit to Proton credit (see DISPATCH-458).
-                //
-                if (link->incremental_credit_CT > 0)
-                    qdr_link_check_credit(core, link);
+                case QDR_LINK_WORK_FLOW :
+                    if (link_work->value > 0)
+                        core->flow_handler(core->user_context, link, link_work->value);
+                    if      (link_work->drain_action == QDR_LINK_WORK_DRAIN_ACTION_SET)
+                        core->drain_handler(core->user_context, link, true);
+                    else if (link_work->drain_action == QDR_LINK_WORK_DRAIN_ACTION_CLEAR)
+                        core->drain_handler(core->user_context, link, false);
+                    else if (link_work->drain_action == QDR_LINK_WORK_DRAIN_ACTION_DRAINED)
+                        core->drained_handler(core->user_context, link);
+                    break;
+
+                case QDR_LINK_WORK_FIRST_DETACH :
+                    core->detach_handler(core->user_context, link, link_work->error, true, link_work->close_link);
+                    break;
+
+                case QDR_LINK_WORK_SECOND_DETACH :
+                    core->detach_handler(core->user_context, link, link_work->error, false, link_work->close_link);
+                    free_qdr_link_t(link);
+                    break;
+                }
+
+                sys_mutex_lock(conn->work_lock);
+                if (link_work->work_type == QDR_LINK_WORK_DELIVERY && link_work->value > 0) {
+                    DEQ_INSERT_HEAD(link->work_list, link_work);
+                    link_work = 0; // Halt work processing
+                } else {
+                    qdr_error_free(link_work->error);
+                    free_qdr_link_work_t(link_work);
+                    link_work = DEQ_HEAD(link->work_list);
+                    if (link_work)
+                        DEQ_REMOVE_HEAD(link->work_list);
+                }
+                sys_mutex_unlock(conn->work_lock);
+                event_count++;
             }
-            if (link->drain_mode_changed) {
-                core->drain_handler(core->user_context, link, link->drain_mode);
-                link->drain_mode_changed = false;
+
+            //
+            // Handle disposition/settlement updates
+            //
+            qdr_delivery_ref_list_t updated_deliveries;
+            sys_mutex_lock(conn->work_lock);
+            DEQ_MOVE(link->updated_deliveries, updated_deliveries);
+            sys_mutex_unlock(conn->work_lock);
+
+            qdr_delivery_ref_t *dref = DEQ_HEAD(updated_deliveries);
+            while (dref) {
+                core->delivery_update_handler(core->user_context, dref->dlv, dref->dlv->disposition, dref->dlv->settled);
+                qdr_delivery_decref(core, dref->dlv);
+                qdr_del_delivery_ref(&updated_deliveries, dref);
+                dref = DEQ_HEAD(updated_deliveries);
+                event_count++;
             }
-            event_count++;
         }
     } while (link);
 
@@ -470,6 +492,21 @@
 }
 
 
+void qdr_link_enqueue_work_CT(qdr_core_t      *core,
+                              qdr_link_t      *link,
+                              qdr_link_work_t *work)
+{
+    qdr_connection_t *conn = link->conn;
+
+    sys_mutex_lock(conn->work_lock);
+    DEQ_INSERT_TAIL(link->work_list, work);
+    qdr_add_link_ref(&conn->links_with_work, link, QDR_LINK_LIST_CLASS_WORK);
+    sys_mutex_unlock(conn->work_lock);
+
+    qdr_connection_activate_CT(core, conn);
+}
+
+
 #define QDR_DISCRIMINATOR_SIZE 16
 static void qdr_generate_discriminator(char *string)
 {
@@ -554,8 +591,10 @@
     qdr_delivery_ref_list_t updated_deliveries;
     qdr_delivery_list_t     undelivered;
     qdr_delivery_list_t     unsettled;
+    qdr_link_work_list_t    work_list;
 
     sys_mutex_lock(conn->work_lock);
+    DEQ_MOVE(link->work_list, work_list);
     DEQ_MOVE(link->updated_deliveries, updated_deliveries);
     DEQ_MOVE(link->undelivered, undelivered);
     qdr_delivery_t *d = DEQ_HEAD(undelivered);
@@ -575,6 +614,17 @@
     sys_mutex_unlock(conn->work_lock);
 
     //
+    // Free the work list
+    //
+    qdr_link_work_t *link_work = DEQ_HEAD(work_list);
+    while (link_work) {
+        DEQ_REMOVE_HEAD(work_list);
+        qdr_error_free(link_work->error);
+        free_qdr_link_work_t(link_work);
+        link_work = DEQ_HEAD(work_list);
+    }
+
+    //
     // Free all the 'updated' references
     //
     qdr_delivery_ref_t *ref = DEQ_HEAD(updated_deliveries);
@@ -675,8 +725,7 @@
     //
     qdr_del_link_ref(&conn->links, link, QDR_LINK_LIST_CLASS_CONNECTION);
     sys_mutex_lock(conn->work_lock);
-    qdr_del_link_ref(&conn->links_with_deliveries, link, QDR_LINK_LIST_CLASS_DELIVERY);
-    qdr_del_link_ref(&conn->links_with_credit,     link, QDR_LINK_LIST_CLASS_FLOW);
+    qdr_del_link_ref(&conn->links_with_work, link, QDR_LINK_LIST_CLASS_WORK);
     sys_mutex_unlock(conn->work_lock);
 
     //
@@ -724,7 +773,6 @@
     work->link      = link;
     work->source    = source;
     work->target    = target;
-    work->error     = 0;
 
     qdr_connection_enqueue_work_CT(core, conn, work);
     return link;
@@ -733,12 +781,10 @@
 
 void qdr_link_outbound_detach_CT(qdr_core_t *core, qdr_link_t *link, qdr_error_t *error, qdr_condition_t condition, bool close)
 {
-    qdr_connection_work_t *work = new_qdr_connection_work_t();
+    qdr_link_work_t *work = new_qdr_link_work_t();
     ZERO(work);
-    work->work_type  = ++link->detach_count == 1 ? QDR_CONNECTION_WORK_FIRST_DETACH : QDR_CONNECTION_WORK_SECOND_DETACH;
-    work->link       = link;
+    work->work_type  = ++link->detach_count == 1 ? QDR_LINK_WORK_FIRST_DETACH : QDR_LINK_WORK_SECOND_DETACH;
     work->close_link = close;
-    work->error      = 0;
 
     if (error)
         work->error = error;
@@ -769,7 +815,7 @@
     if (link->detach_count == 2)
         qdr_link_cleanup_CT(core, link->conn, link);
 
-    qdr_connection_enqueue_work_CT(core, link->conn, work);
+    qdr_link_enqueue_work_CT(core, link, work);
 }
 
 
@@ -781,7 +827,6 @@
     work->link      = link;
     work->source    = source;
     work->target    = target;
-    work->error     = 0;
 
     link->oper_status = QDR_LINK_OPER_UP;
 
@@ -1163,7 +1208,6 @@
         DEQ_REMOVE_HEAD(conn->work_list);
         qdr_terminus_free(work->source);
         qdr_terminus_free(work->target);
-        qdr_error_free(work->error);
         free_qdr_connection_work_t(work);
         work = DEQ_HEAD(conn->work_list);
     }
diff --git a/src/router_core/forwarder.c b/src/router_core/forwarder.c
index debce01..b37b4b8 100644
--- a/src/router_core/forwarder.c
+++ b/src/router_core/forwarder.c
@@ -145,6 +145,19 @@
         if (dlv->settled) {
             DEQ_REMOVE(link->undelivered, dlv);
             dlv->where = QDR_DELIVERY_NOWHERE;
+
+            //
+            // The link-work item representing this pending delivery must be
+            // updated to reflect the removal of the delivery.  If the item
+            // has no other deliveries associated with it, it can be removed
+            // from the work list.
+            //
+            assert(dlv->link_work);
+            if (dlv->link_work && (--dlv->link_work->value == 0)) {
+                DEQ_REMOVE(link->work_list, dlv->link_work);
+                free_qdr_link_work_t(dlv->link_work);
+                dlv->link_work = 0;
+            }
             qdr_delivery_decref_CT(core, dlv);
         }
         dlv = next;
@@ -169,9 +182,22 @@
     qdr_delivery_incref(dlv);
 
     //
-    // If the link isn't already on the links_with_deliveries list, put it there.
+    // We must put a work item on the link's work list to represent this pending delivery.
+    // If there's already a delivery item on the tail of the work list, simply join that item
+    // by incrementing the value.
     //
-    qdr_add_link_ref(&link->conn->links_with_deliveries, link, QDR_LINK_LIST_CLASS_DELIVERY);
+    qdr_link_work_t *work = DEQ_TAIL(link->work_list);
+    if (work && work->work_type == QDR_LINK_WORK_DELIVERY) {
+        work->value++;
+    } else {
+        work = new_qdr_link_work_t();
+        ZERO(work);
+        work->work_type = QDR_LINK_WORK_DELIVERY;
+        work->value     = 1;
+        DEQ_INSERT_TAIL(link->work_list, work);
+        qdr_add_link_ref(&link->conn->links_with_work, link, QDR_LINK_LIST_CLASS_WORK);
+    }
+    dlv->link_work = work;
     sys_mutex_unlock(link->conn->work_lock);
 
     //
@@ -695,7 +721,6 @@
         work->link      = out_link;
         work->source    = source;
         work->target    = target;
-        work->error     = 0;
 
         qdr_connection_enqueue_work_CT(core, conn, work);
 
diff --git a/src/router_core/router_core.c b/src/router_core/router_core.c
index ef73553..3257980 100644
--- a/src/router_core/router_core.c
+++ b/src/router_core/router_core.c
@@ -30,6 +30,7 @@
 ALLOC_DEFINE(qdr_router_ref_t);
 ALLOC_DEFINE(qdr_link_ref_t);
 ALLOC_DEFINE(qdr_general_work_t);
+ALLOC_DEFINE(qdr_link_work_t);
 ALLOC_DEFINE(qdr_connection_ref_t);
 ALLOC_DEFINE(qdr_connection_info_t);
 
diff --git a/src/router_core/router_core_private.h b/src/router_core/router_core_private.h
index c657391..414647b 100644
--- a/src/router_core/router_core_private.h
+++ b/src/router_core/router_core_private.h
@@ -152,6 +152,104 @@
 ALLOC_DECLARE(qdr_action_t);
 DEQ_DECLARE(qdr_action_t, qdr_action_list_t);
 
+//
+// General Work
+//
+// The following types are used to post work to the IO threads for
+// non-connection-specific action.  These actions are serialized through
+// a zero-delay timer and are processed by one thread at a time.  General
+// actions occur in-order and are not run concurrently.
+//
+typedef struct qdr_general_work_t qdr_general_work_t;
+typedef void (*qdr_general_work_handler_t) (qdr_core_t *core, qdr_general_work_t *work);
+
+struct qdr_general_work_t {
+    DEQ_LINKS(qdr_general_work_t);
+    qdr_general_work_handler_t  handler;
+    qdr_field_t                *field;
+    int                         maskbit;
+    int                         inter_router_cost;
+    qdr_receive_t               on_message;
+    void                       *on_message_context;
+    qd_message_t               *msg;
+};
+
+ALLOC_DECLARE(qdr_general_work_t);
+DEQ_DECLARE(qdr_general_work_t, qdr_general_work_list_t);
+
+qdr_general_work_t *qdr_general_work(qdr_general_work_handler_t handler);
+
+
+//
+// Connection Work
+//
+// The following types are used to post work to the IO threads for
+// connection-specific action.  The actions for a particular connection
+// are run in-order and are not concurrent.  Actions for different connections
+// will run concurrently.
+//
+typedef enum {
+    QDR_CONNECTION_WORK_FIRST_ATTACH,
+    QDR_CONNECTION_WORK_SECOND_ATTACH
+} qdr_connection_work_type_t;
+
+typedef struct qdr_connection_work_t {
+    DEQ_LINKS(struct qdr_connection_work_t);
+    qdr_connection_work_type_t  work_type;
+    qdr_link_t                 *link;
+    qdr_terminus_t             *source;
+    qdr_terminus_t             *target;
+} qdr_connection_work_t;
+
+ALLOC_DECLARE(qdr_connection_work_t);
+DEQ_DECLARE(qdr_connection_work_t, qdr_connection_work_list_t);
+
+
+//
+// Link Work
+//
+// The following type is used to post link-specific work to the IO threads.
+// This ensures that work related to a particular link (deliveries, disposition
+// updates, flow updates, and detaches) are processed in-order.
+//
+// DELIVERY      - Push up to _value_ deliveries from the undelivered list to the
+//                 link (outgoing links only).  Don't push more than there is
+//                 available credit for.  If the full number of deliveries (_value_)
+//                 cannot be pushed, don't consume this work item from the list.
+//                 This link will be blocked until further credit is received.
+// FLOW          - Push a flow update using _drain_action_ and _value_ for the
+//                 number of incremental credits.
+// FIRST_DETACH  - Issue a first detach on this link, using _error_ if there is an
+//                 error condition.
+// SECOND_DETACH - Issue a second detach on this link.
+//
+typedef enum {
+    QDR_LINK_WORK_DELIVERY,
+    QDR_LINK_WORK_FLOW,
+    QDR_LINK_WORK_FIRST_DETACH,
+    QDR_LINK_WORK_SECOND_DETACH
+} qdr_link_work_type_t;
+
+typedef enum {
+    QDR_LINK_WORK_DRAIN_ACTION_NONE = 0,
+    QDR_LINK_WORK_DRAIN_ACTION_SET,
+    QDR_LINK_WORK_DRAIN_ACTION_CLEAR,
+    QDR_LINK_WORK_DRAIN_ACTION_DRAINED
+} qdr_link_work_drain_action_t;
+
+typedef struct qdr_link_work_t {
+    DEQ_LINKS(struct qdr_link_work_t);
+    qdr_link_work_type_t          work_type;
+    qdr_error_t                  *error;
+    int                           value;
+    bool                          close_link;
+    qdr_link_work_drain_action_t  drain_action;
+} qdr_link_work_t;
+
+ALLOC_DECLARE(qdr_link_work_t);
+DEQ_DECLARE(qdr_link_work_t, qdr_link_work_list_t);
+
+
 #define QDR_AGENT_MAX_COLUMNS 64
 #define QDR_AGENT_COLUMN_NULL (QDR_AGENT_MAX_COLUMNS + 1)
 
@@ -225,6 +323,7 @@
     qd_bitmask_t        *link_exclusion;
     qdr_address_t       *tracking_addr;
     int                  tracking_addr_bit;
+    qdr_link_work_t     *link_work;         ///< Delivery work item for this delivery
 };
 
 ALLOC_DECLARE(qdr_delivery_t);
@@ -242,10 +341,9 @@
 void qdr_del_delivery_ref(qdr_delivery_ref_list_t *list, qdr_delivery_ref_t *ref);
 
 #define QDR_LINK_LIST_CLASS_ADDRESS    0
-#define QDR_LINK_LIST_CLASS_DELIVERY   1
-#define QDR_LINK_LIST_CLASS_FLOW       2
-#define QDR_LINK_LIST_CLASS_CONNECTION 3
-#define QDR_LINK_LIST_CLASSES          4
+#define QDR_LINK_LIST_CLASS_WORK       1
+#define QDR_LINK_LIST_CLASS_CONNECTION 2
+#define QDR_LINK_LIST_CLASSES          3
 
 typedef enum {
     QDR_LINK_OPER_UP,
@@ -262,6 +360,7 @@
     qdr_connection_t        *conn;               ///< [ref] Connection that owns this link
     qd_link_type_t           link_type;
     qd_direction_t           link_direction;
+    qdr_link_work_list_t     work_list;
     char                    *name;
     int                      detach_count;       ///< 0, 1, or 2 depending on the state of the lifecycle
     qdr_address_t           *owning_addr;        ///< [ref] Address record that owns this link
@@ -276,11 +375,8 @@
     bool                     strip_annotations_in;
     bool                     strip_annotations_out;
     int                      capacity;
-    int                      incremental_credit_CT;
-    int                      incremental_credit;
     bool                     flow_started;   ///< for incoming, true iff initial credit has been granted
     bool                     drain_mode;
-    bool                     drain_mode_changed;
     int                      credit_to_core; ///< Number of the available credits incrementally given to the core
 
     uint64_t total_deliveries;
@@ -389,59 +485,11 @@
 
 
 //
-// General Work
+// Connection Information
 //
-// The following types are used to post work to the IO threads for
-// non-connection-specific action.  These actions are serialized through
-// a zero-delay timer and are processed by one thread at a time.  General
-// actions occur in-order and are not run concurrently.
+// This record is used to give the core thread access to the details
+// of a connection's configuration.
 //
-typedef struct qdr_general_work_t qdr_general_work_t;
-typedef void (*qdr_general_work_handler_t) (qdr_core_t *core, qdr_general_work_t *work);
-
-struct qdr_general_work_t {
-    DEQ_LINKS(qdr_general_work_t);
-    qdr_general_work_handler_t  handler;
-    qdr_field_t                *field;
-    int                         maskbit;
-    int                         inter_router_cost;
-    qdr_receive_t               on_message;
-    void                       *on_message_context;
-    qd_message_t               *msg;
-};
-
-ALLOC_DECLARE(qdr_general_work_t);
-DEQ_DECLARE(qdr_general_work_t, qdr_general_work_list_t);
-
-qdr_general_work_t *qdr_general_work(qdr_general_work_handler_t handler);
-
-//
-// Connection Work
-//
-// The following types are used to post work to the IO threads for
-// connection-specific action.  The actions for a particular connection
-// are run in-order and are not concurrent.  Actions for different connections
-// will run concurrently.
-//
-typedef enum {
-    QDR_CONNECTION_WORK_FIRST_ATTACH,
-    QDR_CONNECTION_WORK_SECOND_ATTACH,
-    QDR_CONNECTION_WORK_FIRST_DETACH,
-    QDR_CONNECTION_WORK_SECOND_DETACH
-} qdr_connection_work_type_t;
-
-typedef struct qdr_connection_work_t {
-    DEQ_LINKS(struct qdr_connection_work_t);
-    qdr_connection_work_type_t  work_type;
-    qdr_link_t                 *link;
-    qdr_terminus_t             *source;
-    qdr_terminus_t             *target;
-    qdr_error_t                *error;
-    bool                        close_link;
-} qdr_connection_work_t;
-
-ALLOC_DECLARE(qdr_connection_work_t);
-DEQ_DECLARE(qdr_connection_work_t, qdr_connection_work_list_t);
 
 struct qdr_connection_info_t {
     const  char                *container;
@@ -481,7 +529,7 @@
     sys_mutex_t                *work_lock;
     qdr_link_ref_list_t         links;
     qdr_link_ref_list_t         links_with_deliveries;
-    qdr_link_ref_list_t         links_with_credit;
+    qdr_link_ref_list_t         links_with_work;
     char                       *tenant_space;
     int                         tenant_space_len;
     qdr_connection_info_t      *connection_info;
@@ -662,6 +710,9 @@
 void qdr_connection_enqueue_work_CT(qdr_core_t            *core,
                                     qdr_connection_t      *conn,
                                     qdr_connection_work_t *work);
+void qdr_link_enqueue_work_CT(qdr_core_t      *core,
+                              qdr_link_t      *conn,
+                              qdr_link_work_t *work);
 
 qdr_link_t *qdr_create_link_CT(qdr_core_t       *core,
                                qdr_connection_t *conn,
@@ -677,10 +728,4 @@
                        qd_router_entity_type_t  type,
                        qd_composed_field_t     *body);
 
-//
-// Cause the core to check credit on an incoming link that might have CT credit but
-// no IO/Proton credit.
-//
-void qdr_link_check_credit(qdr_core_t *core, qdr_link_t *link);
-
 #endif
diff --git a/src/router_core/transfer.c b/src/router_core/transfer.c
index a35e658..66ab3d8 100644
--- a/src/router_core/transfer.c
+++ b/src/router_core/transfer.c
@@ -23,7 +23,6 @@
 
 static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
 static void qdr_link_flow_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
-static void qdr_link_check_credit_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
 static void qdr_send_to_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
 static void qdr_update_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
 static void qdr_delete_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
@@ -129,6 +128,7 @@
             dlv = DEQ_HEAD(link->undelivered);
             if (dlv) {
                 DEQ_REMOVE_HEAD(link->undelivered);
+                dlv->link_work = 0;
                 settled = dlv->settled;
                 if (!settled) {
                     DEQ_INSERT_TAIL(link->unsettled, dlv);
@@ -156,22 +156,6 @@
         else if (offer != -1)
             core->offer_handler(core->user_context, link, offer);
     }
-
-    //
-    // Handle disposition/settlement updates
-    //
-    qdr_delivery_ref_list_t updated_deliveries;
-    sys_mutex_lock(conn->work_lock);
-    DEQ_MOVE(link->updated_deliveries, updated_deliveries);
-    sys_mutex_unlock(conn->work_lock);
-
-    qdr_delivery_ref_t *ref = DEQ_HEAD(updated_deliveries);
-    while (ref) {
-        core->delivery_update_handler(core->user_context, ref->dlv, ref->dlv->disposition, ref->dlv->settled);
-        qdr_delivery_decref(core, ref->dlv);
-        qdr_del_delivery_ref(&updated_deliveries, ref);
-        ref = DEQ_HEAD(updated_deliveries);
-    }
 }
 
 
@@ -197,14 +181,6 @@
 }
 
 
-void qdr_link_check_credit(qdr_core_t *core, qdr_link_t *link)
-{
-    qdr_action_t *action = qdr_action(qdr_link_check_credit_CT, "link_check_credit");
-    action->args.connection.link = link;
-    qdr_action_enqueue(core, action);
-}
-
-
 void qdr_send_to1(qdr_core_t *core, qd_message_t *msg, qd_iterator_t *addr, bool exclude_inprocess, bool control)
 {
     qdr_action_t *action = qdr_action(qdr_send_to_CT, "send_to");
@@ -452,12 +428,13 @@
     if (discard)
         return;
 
-    qdr_link_t *link   = action->args.connection.link;
-    int  credit        = action->args.connection.credit;
-    bool drain         = action->args.connection.drain;
-    bool activate      = false;
-    bool drain_was_set = !link->drain_mode && drain;
-
+    qdr_link_t *link      = action->args.connection.link;
+    int  credit           = action->args.connection.credit;
+    bool drain            = action->args.connection.drain;
+    bool activate         = false;
+    bool drain_was_set    = !link->drain_mode && drain;
+    qdr_link_work_t *work = 0;
+    
     link->drain_mode = drain;
 
     //
@@ -470,10 +447,13 @@
         if (clink->link_direction == QD_INCOMING)
             qdr_link_issue_credit_CT(core, link->connected_link, credit, drain);
         else {
-            sys_mutex_lock(clink->conn->work_lock);
-            qdr_add_link_ref(&clink->conn->links_with_deliveries, clink, QDR_LINK_LIST_CLASS_DELIVERY);
-            sys_mutex_unlock(clink->conn->work_lock);
-            qdr_connection_activate_CT(core, clink->conn);
+            work = new_qdr_link_work_t();
+            ZERO(work);
+            work->work_type = QDR_LINK_WORK_FLOW;
+            work->value     = credit;
+            if (drain)
+                work->drain_action = QDR_LINK_WORK_DRAIN_ACTION_DRAINED;
+            qdr_link_enqueue_work_CT(core, clink, work);
         }
 
         return;
@@ -483,9 +463,18 @@
     // Handle the replenishing of credit outbound
     //
     if (link->link_direction == QD_OUTGOING && (credit > 0 || drain_was_set)) {
+        if (drain_was_set) {
+            work = new_qdr_link_work_t();
+            ZERO(work);
+            work->work_type    = QDR_LINK_WORK_FLOW;
+            work->drain_action = QDR_LINK_WORK_DRAIN_ACTION_DRAINED;
+        }
+
         sys_mutex_lock(link->conn->work_lock);
+        if (work)
+            DEQ_INSERT_TAIL(link->work_list, work);
         if (DEQ_SIZE(link->undelivered) > 0 || drain_was_set) {
-            qdr_add_link_ref(&link->conn->links_with_deliveries, link, QDR_LINK_LIST_CLASS_DELIVERY);
+            qdr_add_link_ref(&link->conn->links_with_work, link, QDR_LINK_LIST_CLASS_WORK);
             activate = true;
         }
         sys_mutex_unlock(link->conn->work_lock);
@@ -499,16 +488,6 @@
 }
 
 
-static void qdr_link_check_credit_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
-{
-    if (discard)
-        return;
-
-    qdr_link_t *link = action->args.connection.link;
-    qdr_link_issue_credit_CT(core, link, 0, false);
-}
-
-
 /**
  * Return the number of outbound paths to destinations that this address has.
  * Note that even if there are more than zero paths, the destination still may
@@ -771,42 +750,31 @@
 
 
 /**
- * Check the link's accumulated credit.  If the credit given to the connection thread
- * has been issued to Proton, provide the next batch of credit to the connection thread.
+ * Add link-work to provide credit to the link in an IO thread
  */
 void qdr_link_issue_credit_CT(qdr_core_t *core, qdr_link_t *link, int credit, bool drain)
 {
+    assert(link->link_direction == QD_INCOMING);
+
     bool drain_changed = link->drain_mode |= drain;
-    bool activate      = drain_changed;
+    link->drain_mode   = drain;
 
-    link->drain_mode = drain;
-    link->drain_mode_changed = drain_changed;
+    if (!drain_changed && credit == 0)
+        return;
 
-    link->incremental_credit_CT += credit;
-    link->flow_started = true;
+    if (credit > 0)
+        link->flow_started = true;
 
-    if (link->incremental_credit_CT && link->incremental_credit == 0) {
-        //
-        // Move the credit from the core-thread value to the connection-thread value.
-        //
-        link->incremental_credit    = link->incremental_credit_CT;
-        link->incremental_credit_CT = 0;
-        activate = true;
-    }
+    qdr_link_work_t *work = new_qdr_link_work_t();
+    ZERO(work);
 
-    if (activate) {
-        //
-        // Put this link on the connection's has-credit list.
-        //
-        sys_mutex_lock(link->conn->work_lock);
-        qdr_add_link_ref(&link->conn->links_with_credit, link, QDR_LINK_LIST_CLASS_FLOW);
-        sys_mutex_unlock(link->conn->work_lock);
+    work->work_type = QDR_LINK_WORK_FLOW;
+    work->value     = credit;
 
-        //
-        // Activate the connection
-        //
-        qdr_connection_activate_CT(core, link->conn);
-    }
+    if (drain_changed)
+        work->drain_action = drain ? QDR_LINK_WORK_DRAIN_ACTION_SET : QDR_LINK_WORK_DRAIN_ACTION_CLEAR;
+
+    qdr_link_enqueue_work_CT(core, link, work);
 }
 
 
@@ -876,7 +844,7 @@
     if (dlv->where != QDR_DELIVERY_IN_UNDELIVERED) {
         qdr_delivery_incref(dlv);
         qdr_add_delivery_ref(&link->updated_deliveries, dlv);
-        qdr_add_link_ref(&link->conn->links_with_deliveries, link, QDR_LINK_LIST_CLASS_DELIVERY);
+        qdr_add_link_ref(&link->conn->links_with_work, link, QDR_LINK_LIST_CLASS_WORK);
         activate = true;
     }
     sys_mutex_unlock(link->conn->work_lock);
diff --git a/src/router_node.c b/src/router_node.c
index ff666e6..5868570 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -991,19 +991,24 @@
 }
 
 
-static void CORE_link_push(void *context, qdr_link_t *link)
+static int CORE_link_push(void *context, qdr_link_t *link, int limit)
 {
     qd_router_t *router = (qd_router_t*) context;
     qd_link_t   *qlink  = (qd_link_t*) qdr_link_get_context(link);
     if (!qlink)
-        return;
+        return 0;
     
     pn_link_t *plink = qd_link_pn(qlink);
 
     if (plink) {
         int link_credit = pn_link_credit(plink);
+        if (link_credit > limit)
+            link_credit = limit;
         qdr_link_process_deliveries(router->router_core, link, link_credit);
+        return link_credit;
     }
+
+    return 0;
 }
 
 
diff --git a/tests/system_tests_drain_support.py b/tests/system_tests_drain_support.py
index e4bd2bc..e59f870 100644
--- a/tests/system_tests_drain_support.py
+++ b/tests/system_tests_drain_support.py
@@ -19,7 +19,7 @@
 
 from proton.handlers import MessagingHandler
 from proton.reactor import Container
-from proton import Message
+from proton import Message, Endpoint
 
 class Timeout(object):
     def __init__(self, parent):
@@ -54,6 +54,22 @@
         self.sender = event.container.create_sender(self.conn, "org.apache.dev")
         self.receiver.flow(1)
 
+    def on_link_flow(self, event):
+        if event.link.is_sender and event.link.credit \
+           and event.link.state & Endpoint.LOCAL_ACTIVE \
+           and event.link.state & Endpoint.REMOTE_ACTIVE :
+            self.on_sendable(event)
+
+        # The fact that the event.link.credit is 0 means that the receiver will not be receiving any more
+        # messages. That along with 10 messages received indicates that the drain worked and we can
+        # declare that the test is successful
+        if self.received_count == 10 and event.link.credit == 0:
+            self.error = None
+            self.timer.cancel()
+            self.receiver.close()
+            self.sender.close()
+            self.conn.close()
+
     def on_sendable(self, event):
         if self.sent_count < 10:
             msg = Message(body="Hello World", properties={'seq': self.sent_count})
@@ -75,16 +91,6 @@
                 # receiver
                 event.receiver.drain(20)
 
-            # The fact that the event.link.credit is 0 means that the receiver will not be receiving any more
-            # messages. That along with 10 messages received indicates that the drain worked and we can
-            # declare that the test is successful
-            if self.received_count == 10 and event.link.credit == 0:
-                self.error = None
-                self.timer.cancel()
-                self.receiver.close()
-                self.sender.close()
-                self.conn.close()
-
     def run(self):
         Container(self).run()