]> git.sur5r.net Git - openldap/blobdiff - servers/slapd/overlays/syncprov.c
ITS#6710
[openldap] / servers / slapd / overlays / syncprov.c
index 2a1c565e02d3ab277c4ee2a49e962e721554480d..7fd374f016aa7b9eecf25400710dd39479c92611 100644 (file)
@@ -2,7 +2,7 @@
 /* syncprov.c - syncrepl provider */
 /* This work is part of OpenLDAP Software <http://www.openldap.org/>.
  *
- * Copyright 2004-2009 The OpenLDAP Foundation.
+ * Copyright 2004-2010 The OpenLDAP Foundation.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -48,6 +48,7 @@ typedef struct modtarget {
 /* A queued result of a persistent search */
 typedef struct syncres {
        struct syncres *s_next;
+       Entry *s_e;
        struct berval s_dn;
        struct berval s_ndn;
        struct berval s_uuid;
@@ -124,6 +125,7 @@ typedef struct sessionlog {
 typedef struct syncprov_info_t {
        syncops         *si_ops;
        BerVarray       si_ctxcsn;      /* ldapsync context */
+       struct berval   si_contextdn;
        int             *si_sids;
        int             si_numcsns;
        int             si_chkops;      /* checkpointing info */
@@ -131,6 +133,9 @@ typedef struct syncprov_info_t {
        int             si_numops;      /* number of ops since last checkpoint */
        int             si_nopres;      /* Skip present phase */
        int             si_usehint;     /* use reload hint */
+       int             si_active;      /* True if there are active mods */
+       int             si_dirty;       /* True if the context is dirty, i.e changes
+                                                * have been made without updating the csn. */
        time_t  si_chklast;     /* time of last checkpoint */
        Avlnode *si_mods;       /* entries being modified */
        sessionlog      *si_logs;
@@ -144,6 +149,7 @@ typedef struct opcookie {
        slap_overinst *son;
        syncmatches *smatches;
        modtarget *smt;
+       Entry *se;
        struct berval sdn;      /* DN of entry, for deletes */
        struct berval sndn;
        struct berval suuid;    /* UUID of entry */
@@ -153,6 +159,11 @@ typedef struct opcookie {
        short sreference;       /* Is the entry a reference? */
 } opcookie;
 
+typedef struct mutexint {
+       ldap_pvt_thread_mutex_t mi_mutex;
+       int mi_int;
+} mutexint;
+
 typedef struct fbase_cookie {
        struct berval *fdn;     /* DN of a modified entry, for scope testing */
        syncops *fss;   /* persistent search we're testing against */
@@ -583,8 +594,8 @@ syncprov_findcsn( Operation *op, find_csn_t mode )
        slap_callback cb = {0};
        Operation fop;
        SlapReply frs = { REP_RESULT };
-       char buf[LDAP_LUTIL_CSNSTR_BUFSIZE + STRLENOF("(entryCSN<=)")];
-       char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
+       char buf[LDAP_PVT_CSNSTR_BUFSIZE + STRLENOF("(entryCSN<=)")];
+       char cbuf[LDAP_PVT_CSNSTR_BUFSIZE];
        struct berval maxcsn;
        Filter cf;
        AttributeAssertion eq = ATTRIBUTEASSERTION_INIT;
@@ -738,6 +749,36 @@ again:
        return rc;
 }
 
+/* Should find a place to cache these */
+static mutexint *get_mutexint()
+{
+       mutexint *mi = ch_malloc( sizeof( mutexint ));
+       ldap_pvt_thread_mutex_init( &mi->mi_mutex );
+       mi->mi_int = 1;
+       return mi;
+}
+
+static void inc_mutexint( mutexint *mi )
+{
+       ldap_pvt_thread_mutex_lock( &mi->mi_mutex );
+       mi->mi_int++;
+       ldap_pvt_thread_mutex_unlock( &mi->mi_mutex );
+}
+
+/* return resulting counter */
+static int dec_mutexint( mutexint *mi )
+{
+       int i;
+       ldap_pvt_thread_mutex_lock( &mi->mi_mutex );
+       i = --mi->mi_int;
+       ldap_pvt_thread_mutex_unlock( &mi->mi_mutex );
+       if ( !i ) {
+               ldap_pvt_thread_mutex_destroy( &mi->mi_mutex );
+               ch_free( mi );
+       }
+       return i;
+}
+
 static void
 syncprov_free_syncop( syncops *so )
 {
@@ -761,6 +802,12 @@ syncprov_free_syncop( syncops *so )
        ch_free( so->s_base.bv_val );
        for ( sr=so->s_res; sr; sr=srnext ) {
                srnext = sr->s_next;
+               if ( sr->s_e ) {
+                       if ( !dec_mutexint( sr->s_e->e_private )) {
+                               sr->s_e->e_private = NULL;
+                               entry_free( sr->s_e );
+                       }
+               }
                ch_free( sr );
        }
        ldap_pvt_thread_mutex_destroy( &so->s_mutex );
@@ -769,8 +816,7 @@ syncprov_free_syncop( syncops *so )
 
 /* Send a persistent search response */
 static int
-syncprov_sendresp( Operation *op, opcookie *opc, syncops *so,
-       Entry **e, int mode )
+syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, int mode )
 {
        slap_overinst *on = opc->son;
 
@@ -791,15 +837,13 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so,
        }
 
 #ifdef LDAP_DEBUG
-       if ( !BER_BVISNULL( &cookie )) {
-               if ( so->s_sid > 0 ) {
-                       Debug( LDAP_DEBUG_SYNC, "syncprov_sendresp: to=%03x, cookie=%s\n",
-                               so->s_sid, cookie.bv_val , 0 );
-               } else {
-                       Debug( LDAP_DEBUG_SYNC, "syncprov_sendresp: cookie=%s\n",
-                               cookie.bv_val, 0, 0 );
-               }
-       }               
+       if ( so->s_sid > 0 ) {
+               Debug( LDAP_DEBUG_SYNC, "syncprov_sendresp: to=%03x, cookie=%s\n",
+                       so->s_sid, cookie.bv_val ? cookie.bv_val : "", 0 );
+       } else {
+               Debug( LDAP_DEBUG_SYNC, "syncprov_sendresp: cookie=%s\n",
+                       cookie.bv_val ? cookie.bv_val : "", 0, 0 );
+       }
 #endif
 
        e_uuid.e_attrs = &a_uuid;
@@ -812,35 +856,29 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so,
        }
 
        rs.sr_ctrls = ctrls;
-       op->o_bd->bd_info = (BackendInfo *)on->on_info;
+       rs.sr_entry = &e_uuid;
+       if ( mode == LDAP_SYNC_ADD || mode == LDAP_SYNC_MODIFY ) {
+               e_uuid = *opc->se;
+               e_uuid.e_private = NULL;
+       }
+
        switch( mode ) {
        case LDAP_SYNC_ADD:
-               rs.sr_entry = *e;
-               if ( rs.sr_entry->e_private )
-                       rs.sr_flags = REP_ENTRY_MUSTRELEASE;
                if ( opc->sreference && so->s_op->o_managedsait <= SLAP_CONTROL_IGNORED ) {
                        rs.sr_ref = get_entry_referrals( op, rs.sr_entry );
                        rs.sr_err = send_search_reference( op, &rs );
                        ber_bvarray_free( rs.sr_ref );
-                       if ( !rs.sr_entry )
-                               *e = NULL;
                        break;
                }
                /* fallthru */
        case LDAP_SYNC_MODIFY:
-               rs.sr_entry = *e;
-               if ( rs.sr_entry->e_private )
-                       rs.sr_flags = REP_ENTRY_MUSTRELEASE;
                rs.sr_attrs = op->ors_attrs;
                rs.sr_err = send_search_entry( op, &rs );
-               if ( !rs.sr_entry )
-                       *e = NULL;
                break;
        case LDAP_SYNC_DELETE:
                e_uuid.e_attrs = NULL;
                e_uuid.e_name = opc->sdn;
                e_uuid.e_nname = opc->sndn;
-               rs.sr_entry = &e_uuid;
                if ( opc->sreference && so->s_op->o_managedsait <= SLAP_CONTROL_IGNORED ) {
                        struct berval bv = BER_BVNULL;
                        rs.sr_ref = &bv;
@@ -854,7 +892,14 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so,
        }
        /* In case someone else freed it already? */
        if ( rs.sr_ctrls ) {
-               op->o_tmpfree( rs.sr_ctrls[0], op->o_tmpmemctx );
+               int i;
+               for ( i=0; rs.sr_ctrls[i]; i++ ) {
+                       if ( rs.sr_ctrls[i] == ctrls[0] ) {
+                               op->o_tmpfree( ctrls[0]->ldctl_value.bv_val, op->o_tmpmemctx );
+                               ctrls[0]->ldctl_value.bv_val = NULL;
+                               break;
+                       }
+               }
                rs.sr_ctrls = NULL;
        }
 
@@ -889,9 +934,9 @@ syncprov_qplay( Operation *op, syncops *so )
                ldap_pvt_thread_mutex_unlock( &so->s_mutex );
 
                if ( sr->s_mode == LDAP_SYNC_NEW_COOKIE ) {
-                   SlapReply rs = { REP_INTERMEDIATE };
+                       SlapReply rs = { REP_INTERMEDIATE };
 
-                   rc = syncprov_sendinfo( op, &rs, LDAP_TAG_SYNC_NEW_COOKIE,
+                       rc = syncprov_sendinfo( op, &rs, LDAP_TAG_SYNC_NEW_COOKIE,
                                &sr->s_csn, 0, NULL, 0 );
                } else {
                        opc.sdn = sr->s_dn;
@@ -899,22 +944,15 @@ syncprov_qplay( Operation *op, syncops *so )
                        opc.suuid = sr->s_uuid;
                        opc.sctxcsn = sr->s_csn;
                        opc.sreference = sr->s_isreference;
-                       e = NULL;
-
-                       if ( sr->s_mode != LDAP_SYNC_DELETE ) {
-                               rc = overlay_entry_get_ov( op, &opc.sndn, NULL, NULL, 0, &e, on );
-                               if ( rc ) {
-                                       Debug( LDAP_DEBUG_SYNC, "syncprov_qplay: failed to get %s, "
-                                               "error (%d), ignoring...\n", opc.sndn.bv_val, rc, 0 );
-                                       ch_free( sr );
-                                       rc = 0;
-                                       continue;
-                               }
-                       }
-                       rc = syncprov_sendresp( op, &opc, so, &e, sr->s_mode );
+                       opc.se = sr->s_e;
 
-                       if ( e ) {
-                               overlay_entry_release_ov( op, e, 0, on );
+                       rc = syncprov_sendresp( op, &opc, so, sr->s_mode );
+
+               }
+               if ( sr->s_e ) {
+                       if ( !dec_mutexint( sr->s_e->e_private )) {
+                               sr->s_e->e_private = NULL;
+                               entry_free ( sr->s_e );
                        }
                }
 
@@ -1008,6 +1046,10 @@ syncprov_qresp( opcookie *opc, syncops *so, int mode )
                srsize += cookie.bv_len + 1;
        sr = ch_malloc( srsize );
        sr->s_next = NULL;
+       sr->s_e = opc->se;
+       /* bump refcount on this entry */
+       if ( opc->se )
+               inc_mutexint( opc->se->e_private );
        sr->s_dn.bv_val = (char *)(sr + 1);
        sr->s_dn.bv_len = opc->sdn.bv_len;
        sr->s_mode = mode;
@@ -1151,9 +1193,12 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
                rc = overlay_entry_get_ov( op, fc.fdn, NULL, NULL, 0, &e, on );
                /* If we're sending responses now, make a copy and unlock the DB */
                if ( e && !saveit ) {
-                       Entry *e2 = entry_dup( e );
+                       if ( !opc->se ) {
+                               opc->se = entry_dup( e );
+                               opc->se->e_private = get_mutexint();
+                       }
                        overlay_entry_release_ov( op, e, 0, on );
-                       e = e2;
+                       e = opc->se;
                }
                if ( rc ) {
                        op->o_bd = b0;
@@ -1161,6 +1206,13 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
                }
        } else {
                e = op->ora_e;
+               if ( !saveit ) {
+                       if ( !opc->se ) {
+                               opc->se = entry_dup( e );
+                               opc->se->e_private = get_mutexint();
+                       }
+                       e = opc->se;
+               }
        }
 
        if ( saveit || op->o_tag == LDAP_REQ_ADD ) {
@@ -1190,22 +1242,18 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
                if ( ss->s_op->o_abandon )
                        continue;
 
-               /* First time thru, check for possible skips */
-               if ( saveit || op->o_tag == LDAP_REQ_ADD ) {
-
-                       /* Don't send ops back to the originator */
-                       if ( opc->osid > 0 && opc->osid == ss->s_sid ) {
-                               Debug( LDAP_DEBUG_SYNC, "syncprov_matchops: skipping original sid %03x\n",
-                                       opc->osid, 0, 0 );
-                               continue;
-                       }
+               /* Don't send ops back to the originator */
+               if ( opc->osid > 0 && opc->osid == ss->s_sid ) {
+                       Debug( LDAP_DEBUG_SYNC, "syncprov_matchops: skipping original sid %03x\n",
+                               opc->osid, 0, 0 );
+                       continue;
+               }
 
-                       /* Don't send ops back to the messenger */
-                       if ( opc->rsid > 0 && opc->rsid == ss->s_sid ) {
-                               Debug( LDAP_DEBUG_SYNC, "syncprov_matchops: skipping relayed sid %03x\n",
-                                       opc->rsid, 0, 0 );
-                               continue;
-                       }
+               /* Don't send ops back to the messenger */
+               if ( opc->rsid > 0 && opc->rsid == ss->s_sid ) {
+                       Debug( LDAP_DEBUG_SYNC, "syncprov_matchops: skipping relayed sid %03x\n",
+                               opc->rsid, 0, 0 );
+                       continue;
                }
 
                /* validate base */
@@ -1225,7 +1273,6 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
                        continue;
                }
 
-
                /* If we're sending results now, look for this op in old matches */
                if ( !saveit ) {
                        syncmatches *old;
@@ -1249,6 +1296,7 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
                }
 
                if ( fc.fscope ) {
+                       ldap_pvt_thread_mutex_lock( &ss->s_mutex );
                        op2 = *ss->s_op;
                        oh = *op->o_hdr;
                        oh.oh_conn = ss->s_op->o_conn;
@@ -1257,7 +1305,14 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
                        op2.o_hdr = &oh;
                        op2.o_extra = op->o_extra;
                        op2.o_callback = NULL;
-                       rc = test_filter( &op2, e, ss->s_op->ors_filter );
+                       if (ss->s_flags & PS_FIX_FILTER) {
+                               /* Skip the AND/GE clause that we stuck on in front. We
+                                  would lose deletes/mods that happen during the refresh
+                                  phase otherwise (ITS#6555) */
+                               op2.ors_filter = ss->s_op->ors_filter->f_and->f_next;
+                       }
+                       ldap_pvt_thread_mutex_unlock( &ss->s_mutex );
+                       rc = test_filter( &op2, e, op2.ors_filter );
                }
 
                Debug( LDAP_DEBUG_TRACE, "syncprov_matchops: sid %03x fscope %d rc %d\n",
@@ -1297,9 +1352,17 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
                if ( !SLAP_ISOVERLAY( op->o_bd )) {
                        op->o_bd = &db;
                }
-               overlay_entry_release_ov( op, e, 0, on );
+               if ( saveit )
+                       overlay_entry_release_ov( op, e, 0, on );
                op->o_bd = b0;
        }
+       if ( opc->se && !saveit ) {
+               if ( !dec_mutexint( opc->se->e_private )) {
+                       opc->se->e_private = NULL;
+                       entry_free( opc->se );
+                       opc->se = NULL;
+               }
+       }
        if ( freefdn ) {
                op->o_tmpfree( fc.fdn->bv_val, op->o_tmpmemctx );
        }
@@ -1316,6 +1379,11 @@ syncprov_op_cleanup( Operation *op, SlapReply *rs )
        syncmatches *sm, *snext;
        modtarget *mt, mtdummy;
 
+       ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
+       if ( si->si_active )
+               si->si_active--;
+       ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
+
        for (sm = opc->smatches; sm; sm=snext) {
                snext = sm->sm_next;
                syncprov_free_syncop( sm->sm_op );
@@ -1361,6 +1429,8 @@ syncprov_checkpoint( Operation *op, SlapReply *rs, slap_overinst *on )
        SlapReply rsm = { 0 };
        slap_callback cb = {0};
        BackendDB be;
+       BackendInfo *bi;
+
 #ifdef CHECK_CSN
        Syntax *syn = slap_schema.si_ad_contextCSN->ad_type->sat_syntax;
 
@@ -1387,12 +1457,28 @@ syncprov_checkpoint( Operation *op, SlapReply *rs, slap_overinst *on )
                be = *on->on_info->oi_origdb;
                opm.o_bd = &be;
        }
-       opm.o_req_dn = opm.o_bd->be_suffix[0];
-       opm.o_req_ndn = opm.o_bd->be_nsuffix[0];
+       opm.o_req_dn = si->si_contextdn;
+       opm.o_req_ndn = si->si_contextdn;
+       bi = opm.o_bd->bd_info;
        opm.o_bd->bd_info = on->on_info->oi_orig;
        opm.o_managedsait = SLAP_CONTROL_NONCRITICAL;
        opm.o_no_schema_check = 1;
        opm.o_bd->be_modify( &opm, &rsm );
+
+       if ( rsm.sr_err == LDAP_NO_SUCH_OBJECT &&
+               SLAP_SYNC_SUBENTRY( opm.o_bd )) {
+               const char      *text;
+               char txtbuf[SLAP_TEXT_BUFLEN];
+               size_t textlen = sizeof txtbuf;
+               Entry *e = slap_create_context_csn_entry( opm.o_bd, NULL );
+               slap_mods2entry( &mod, &e, 0, 1, &text, txtbuf, textlen);
+               opm.ora_e = e;
+               opm.o_bd->be_add( &opm, &rsm );
+               if ( e == opm.ora_e )
+                       be_entry_release_w( &opm, opm.ora_e );
+       }
+       opm.o_bd->bd_info = bi;
+
        if ( mod.sml_next != NULL ) {
                slap_mods_free( mod.sml_next, 1 );
        }
@@ -1468,7 +1554,7 @@ syncprov_playlog( Operation *op, SlapReply *rs, sessionlog *sl,
        slap_overinst           *on = (slap_overinst *)op->o_bd->bd_info;
        slog_entry *se;
        int i, j, ndel, num, nmods, mmods;
-       char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
+       char cbuf[LDAP_PVT_CSNSTR_BUFSIZE];
        BerVarray uuids;
        struct berval delcsn[2];
 
@@ -1643,7 +1729,7 @@ syncprov_op_response( Operation *op, SlapReply *rs )
        if ( rs->sr_err == LDAP_SUCCESS )
        {
                struct berval maxcsn;
-               char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
+               char cbuf[LDAP_PVT_CSNSTR_BUFSIZE];
                int do_check = 0, have_psearches, foundit, csn_changed = 0;
 
                ldap_pvt_thread_mutex_lock( &si->si_resp_mutex );
@@ -1654,7 +1740,50 @@ syncprov_op_response( Operation *op, SlapReply *rs )
                maxcsn.bv_len = sizeof(cbuf);
                ldap_pvt_thread_rdwr_wlock( &si->si_csn_rwlock );
 
-               if ( op->o_dont_replicate && op->o_tag == LDAP_REQ_MODIFY &&
+               slap_get_commit_csn( op, &maxcsn, &foundit );
+               if ( BER_BVISEMPTY( &maxcsn ) && SLAP_GLUE_SUBORDINATE( op->o_bd )) {
+                       /* syncrepl queues the CSN values in the db where
+                        * it is configured , not where the changes are made.
+                        * So look for a value in the glue db if we didn't
+                        * find any in this db.
+                        */
+                       BackendDB *be = op->o_bd;
+                       op->o_bd = select_backend( &be->be_nsuffix[0], 1);
+                       maxcsn.bv_val = cbuf;
+                       maxcsn.bv_len = sizeof(cbuf);
+                       slap_get_commit_csn( op, &maxcsn, &foundit );
+                       op->o_bd = be;
+               }
+               if ( !BER_BVISEMPTY( &maxcsn ) ) {
+                       int i, sid;
+#ifdef CHECK_CSN
+                       Syntax *syn = slap_schema.si_ad_contextCSN->ad_type->sat_syntax;
+                       assert( !syn->ssyn_validate( syn, &maxcsn ));
+#endif
+                       sid = slap_parse_csn_sid( &maxcsn );
+                       for ( i=0; i<si->si_numcsns; i++ ) {
+                               if ( sid == si->si_sids[i] ) {
+                                       if ( ber_bvcmp( &maxcsn, &si->si_ctxcsn[i] ) > 0 ) {
+                                               ber_bvreplace( &si->si_ctxcsn[i], &maxcsn );
+                                               csn_changed = 1;
+                                       }
+                                       break;
+                               }
+                       }
+                       /* It's a new SID for us */
+                       if ( i == si->si_numcsns ) {
+                               value_add_one( &si->si_ctxcsn, &maxcsn );
+                               csn_changed = 1;
+                               si->si_numcsns++;
+                               si->si_sids = ch_realloc( si->si_sids, si->si_numcsns *
+                                       sizeof(int));
+                               si->si_sids[i] = sid;
+                       }
+               }
+
+               /* Don't do any processing for consumer contextCSN updates */
+               if ( op->o_dont_replicate ) {
+                       if ( op->o_tag == LDAP_REQ_MODIFY &&
                                op->orm_modlist->sml_op == LDAP_MOD_REPLACE &&
                                op->orm_modlist->sml_desc == slap_schema.si_ad_contextCSN ) {
                        /* Catch contextCSN updates from syncrepl. We have to look at
@@ -1666,7 +1795,6 @@ syncprov_op_response( Operation *op, SlapReply *rs )
 
                        for ( i=0; i<mod->sml_numvals; i++ ) {
                                sid = slap_parse_csn_sid( &mod->sml_values[i] );
-
                                for ( j=0; j<si->si_numcsns; j++ ) {
                                        if ( sid == si->si_sids[j] ) {
                                                if ( ber_bvcmp( &mod->sml_values[i], &si->si_ctxcsn[j] ) > 0 ) {
@@ -1686,6 +1814,8 @@ syncprov_op_response( Operation *op, SlapReply *rs )
                                        csn_changed = 1;
                                }
                        }
+                       if ( csn_changed )
+                               si->si_dirty = 0;
                        ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
 
                        if ( csn_changed ) {
@@ -1701,59 +1831,9 @@ syncprov_op_response( Operation *op, SlapReply *rs )
                                        }
                                }
                        }
-                       goto leave;
-               }
-
-               slap_get_commit_csn( op, &maxcsn, &foundit );
-               if ( BER_BVISEMPTY( &maxcsn ) && SLAP_GLUE_SUBORDINATE( op->o_bd )) {
-                       /* syncrepl queues the CSN values in the db where
-                        * it is configured , not where the changes are made.
-                        * So look for a value in the glue db if we didn't
-                        * find any in this db.
-                        */
-                       BackendDB *be = op->o_bd;
-                       op->o_bd = select_backend( &be->be_nsuffix[0], 1);
-                       maxcsn.bv_val = cbuf;
-                       maxcsn.bv_len = sizeof(cbuf);
-                       slap_get_commit_csn( op, &maxcsn, &foundit );
-                       op->o_bd = be;
-               }
-               if ( !BER_BVISEMPTY( &maxcsn ) ) {
-                       int i, sid;
-#ifdef CHECK_CSN
-                       Syntax *syn = slap_schema.si_ad_contextCSN->ad_type->sat_syntax;
-                       assert( !syn->ssyn_validate( syn, &maxcsn ));
-#endif
-                       sid = slap_parse_csn_sid( &maxcsn );
-                       for ( i=0; i<si->si_numcsns; i++ ) {
-                               if ( sid == si->si_sids[i] ) {
-                                       if ( ber_bvcmp( &maxcsn, &si->si_ctxcsn[i] ) > 0 ) {
-                                               ber_bvreplace( &si->si_ctxcsn[i], &maxcsn );
-                                               csn_changed = 1;
-                                       }
-                                       break;
-                               }
-                       }
-                       /* It's a new SID for us */
-                       if ( i == si->si_numcsns ) {
-                               value_add_one( &si->si_ctxcsn, &maxcsn );
-                               csn_changed = 1;
-                               si->si_numcsns++;
-                               si->si_sids = ch_realloc( si->si_sids, si->si_numcsns *
-                                       sizeof(int));
-                               si->si_sids[i] = sid;
-                       }
-#if 0
-               } else if ( !foundit ) {
-                       /* internal ops that aren't meant to be replicated */
-                       ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
-                       return SLAP_CB_CONTINUE;
-#endif
-               }
-
-               /* Don't do any processing for consumer contextCSN updates */
-               if ( op->o_dont_replicate ) {
+                       } else {
                        ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
+                       }
                        goto leave;
                }
 
@@ -1763,7 +1843,7 @@ syncprov_op_response( Operation *op, SlapReply *rs )
                         * it will deadlock
                         */
                        if ( op->o_tag != LDAP_REQ_ADD ||
-                               !dn_match( &op->o_req_ndn, &op->o_bd->be_nsuffix[0] )) {
+                               !dn_match( &op->o_req_ndn, &si->si_contextdn )) {
                                if ( si->si_chkops && si->si_numops >= si->si_chkops ) {
                                        do_check = 1;
                                        si->si_numops = 0;
@@ -1779,6 +1859,7 @@ syncprov_op_response( Operation *op, SlapReply *rs )
                                }
                        }
                }
+               si->si_dirty = !csn_changed;
                ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
 
                if ( do_check ) {
@@ -1837,14 +1918,14 @@ syncprov_op_compare( Operation *op, SlapReply *rs )
        syncprov_info_t         *si = on->on_bi.bi_private;
        int rc = SLAP_CB_CONTINUE;
 
-       if ( dn_match( &op->o_req_ndn, op->o_bd->be_nsuffix ) &&
+       if ( dn_match( &op->o_req_ndn, &si->si_contextdn ) &&
                op->oq_compare.rs_ava->aa_desc == slap_schema.si_ad_contextCSN )
        {
                Entry e = {0};
                Attribute a = {0};
 
-               e.e_name = op->o_bd->be_suffix[0];
-               e.e_nname = op->o_bd->be_nsuffix[0];
+               e.e_name = si->si_contextdn;
+               e.e_nname = si->si_contextdn;
                e.e_attrs = &a;
 
                a.a_desc = slap_schema.si_ad_contextCSN;
@@ -1906,6 +1987,7 @@ syncprov_op_mod( Operation *op, SlapReply *rs )
 
        ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
        have_psearches = ( si->si_ops != NULL );
+       si->si_active++;
        ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
 
        cbsize = sizeof(slap_callback) + sizeof(opcookie) +
@@ -2296,6 +2378,7 @@ syncprov_op_search( Operation *op, SlapReply *rs )
        BerVarray ctxcsn;
        int i, *sids, numcsns;
        struct berval mincsn;
+       int dirty = 0;
 
        if ( !(op->o_sync_mode & SLAP_SYNC_REFRESH) ) return SLAP_CB_CONTINUE;
 
@@ -2340,6 +2423,20 @@ syncprov_op_search( Operation *op, SlapReply *rs )
                sop->s_inuse = 1;
 
                ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
+               while ( si->si_active ) {
+                       /* Wait for active mods to finish before proceeding, as they
+                        * may already have inspected the si_ops list looking for
+                        * consumers to replicate the change to.  Using the log
+                        * doesn't help, as we may finish playing it before the
+                        * active mods gets added to it.
+                        */
+                       ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
+                       if ( slapd_shutdown )
+                               return SLAPD_ABANDON;
+                       if ( !ldap_pvt_thread_pool_pausecheck( &connection_pool ))
+                               ldap_pvt_thread_yield();
+                       ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
+               }
                sop->s_next = si->si_ops;
                si->si_ops = sop;
                ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
@@ -2357,6 +2454,7 @@ syncprov_op_search( Operation *op, SlapReply *rs )
                ctxcsn = NULL;
                sids = NULL;
        }
+       dirty = si->si_dirty;
        ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock );
        
        /* If we have a cookie, handle the PRESENT lookups */
@@ -2413,7 +2511,10 @@ syncprov_op_search( Operation *op, SlapReply *rs )
                                        if ( newer < 0 )
                                                changed = SS_CHANGED;
                                        else if ( newer > 0 ) {
-                                       /* our state is older, tell consumer nothing */
+                                       /* our state is older, complain to consumer */
+                                               rs->sr_err = LDAP_UNWILLING_TO_PERFORM;
+                                               rs->sr_text = "consumer state is newer than provider!";
+bailout:
                                                if ( sop ) {
                                                        syncops **sp = &si->si_ops;
                                                        
@@ -2424,7 +2525,6 @@ syncprov_op_search( Operation *op, SlapReply *rs )
                                                        ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
                                                        ch_free( sop );
                                                }
-                                               rs->sr_err = LDAP_SUCCESS;
                                                rs->sr_ctrls = NULL;
                                                send_ldap_result( op, rs );
                                                return rs->sr_err;
@@ -2434,7 +2534,7 @@ syncprov_op_search( Operation *op, SlapReply *rs )
                                if ( changed )
                                        break;
                        }
-                       if ( !changed ) {
+                       if ( !changed && !dirty ) {
                                do_present = 0;
 no_change:             if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) {
                                        LDAPControl     *ctrls[2];
@@ -2479,8 +2579,9 @@ no_change:                if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) {
                                        ber_bvarray_free_x( ctxcsn, op->o_tmpmemctx );
                                if ( sids )
                                        op->o_tmpfree( sids, op->o_tmpmemctx );
-                               send_ldap_error( op, rs, LDAP_SYNC_REFRESH_REQUIRED, "sync cookie is stale" );
-                               return rs->sr_err;
+                               rs->sr_err = LDAP_SYNC_REFRESH_REQUIRED;
+                               rs->sr_text = "sync cookie is stale";
+                               goto bailout;
                        }
                        if ( srs->sr_state.ctxcsn ) {
                                ber_bvarray_free_x( srs->sr_state.ctxcsn, op->o_tmpmemctx );
@@ -2500,8 +2601,7 @@ no_change:                if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) {
                                        ber_bvarray_free_x( ctxcsn, op->o_tmpmemctx );
                                if ( sids )
                                        op->o_tmpfree( sids, op->o_tmpmemctx );
-                               send_ldap_result( op, rs );
-                               return rs->sr_err;
+                               goto bailout;
                        }
                }
        } else {
@@ -2518,7 +2618,7 @@ shortcut:
        }
 
        /* If something changed, find the changes */
-       if ( gotstate && changed ) {
+       if ( gotstate && ( changed || dirty ) ) {
                Filter *fand, *fava;
 
                fand = op->o_tmpalloc( sizeof(Filter), op->o_tmpmemctx );
@@ -2534,10 +2634,14 @@ shortcut:
 #endif
                ber_dupbv_x( &fava->f_ava->aa_value, &mincsn, op->o_tmpmemctx );
                fava->f_next = op->ors_filter;
+               if ( sop )
+                       ldap_pvt_thread_mutex_lock( &sop->s_mutex );
                op->ors_filter = fand;
                filter2bv_x( op, op->ors_filter, &op->ors_filterstr );
-               if ( sop )
+               if ( sop ) {
                        sop->s_flags |= PS_FIX_FILTER;
+                       ldap_pvt_thread_mutex_unlock( &sop->s_mutex );
+               }
        }
 
        /* Let our callback add needed info to returned entries */
@@ -2559,7 +2663,7 @@ shortcut:
         * the refresh phase, just invoke the response callback to transition
         * us into persist phase
         */
-       if ( !changed ) {
+       if ( !changed && !dirty ) {
                rs->sr_err = LDAP_SUCCESS;
                rs->sr_nentries = 0;
                send_ldap_result( op, rs );
@@ -2583,7 +2687,7 @@ syncprov_operational(
                return SLAP_CB_CONTINUE;
 
        if ( rs->sr_entry &&
-               dn_match( &rs->sr_entry->e_nname, op->o_bd->be_nsuffix )) {
+               dn_match( &rs->sr_entry->e_nname, &si->si_contextdn )) {
 
                if ( SLAP_OPATTRS( rs->sr_attr_flags ) ||
                        ad_inlist( slap_schema.si_ad_contextCSN, rs->sr_attrs )) {
@@ -2800,7 +2904,7 @@ sp_cf_gen(ConfigArgs *c)
                }
                sl = si->si_logs;
                if ( !sl ) {
-                       sl = ch_malloc( sizeof( sessionlog ) + LDAP_LUTIL_CSNSTR_BUFSIZE );
+                       sl = ch_malloc( sizeof( sessionlog ) + LDAP_PVT_CSNSTR_BUFSIZE );
                        sl->sl_mincsn.bv_val = (char *)(sl+1);
                        sl->sl_mincsn.bv_len = 0;
                        sl->sl_num = 0;
@@ -2870,13 +2974,19 @@ syncprov_db_open(
        }
 
        thrctx = ldap_pvt_thread_pool_context();
-       connection_fake_init( &conn, &opbuf, thrctx );
+       connection_fake_init2( &conn, &opbuf, thrctx, 0 );
        op = &opbuf.ob_op;
        op->o_bd = be;
        op->o_dn = be->be_rootdn;
        op->o_ndn = be->be_rootndn;
 
-       rc = overlay_entry_get_ov( op, be->be_nsuffix, NULL,
+       if ( SLAP_SYNC_SUBENTRY( be )) {
+               build_new_dn( &si->si_contextdn, be->be_nsuffix,
+                       (struct berval *)&slap_ldapsync_cn_bv, NULL );
+       } else {
+               si->si_contextdn = be->be_nsuffix[0];
+       }
+       rc = overlay_entry_get_ov( op, &si->si_contextdn, NULL,
                slap_schema.si_ad_contextCSN, 0, &e, on );
 
        if ( e ) {
@@ -2900,7 +3010,7 @@ syncprov_db_open(
 
        /* Didn't find a contextCSN, should we generate one? */
        if ( !si->si_ctxcsn ) {
-               char csnbuf[ LDAP_LUTIL_CSNSTR_BUFSIZE ];
+               char csnbuf[ LDAP_PVT_CSNSTR_BUFSIZE ];
                struct berval csn;
 
                if ( SLAP_SYNC_SHADOW( op->o_bd )) {
@@ -2949,7 +3059,7 @@ syncprov_db_close(
                void *thrctx;
 
                thrctx = ldap_pvt_thread_pool_context();
-               connection_fake_init( &conn, &opbuf, thrctx );
+               connection_fake_init2( &conn, &opbuf, thrctx, 0 );
                op = &opbuf.ob_op;
                op->o_bd = be;
                op->o_dn = be->be_rootdn;