Ticket 47819 - Improve tombstone purging performance

Precise Tombstone Purging

A new attribute was added to the nsTombstone objectclass: nsTombstoneCSN,
and a new system equality index was created for nsTombstoneCSN.  When
enabled the tombstone purging thread will use a range search to retrieve
all the tombstones that need to be purged.  Previously we just searched
on "objectclass=nsTombstone", which can be very inefficient if there are
thousands of tombstones, but only a few tombstones that actually need to
be purged.

Migration and Mixed Environments:

Whether or not precise tombstone purging is enabled, the new index is maintained.
The only thing that needs to be done prior to enabling precise purging is
to re-import the database, or run the fixup task.  During replication any
replicated tombstone that is missing the nsTombstoneCSN attribute will have
it added.  During an import, any tombstone missing nsTombstoneCSN will also
have it added to the entry.

Fixup Task:

Specific backend names or suffix DN's can be specified.  If no backend or
suffix is specified then all backends are processed.

    dn: cn=fixem,cn=fixup tombstones,cn=tasks,cn=config
    objectclass: top
    objectclass: extensibleObject
    cn: fixem
    backend: userRoot
    suffix: dc=redhat,dc=com
    suffix: o=test.com
    stripcsn: yes   --> hidden option used for verifying the fixup tasks
                        works(this is used by lib389)

Upgrade:

Added 50nstombstonecsn.ldif which adds nsTombstoneCSN to the system indexes
during an upgrade

Tombstone Purging:

New configuration setting in replica entry:

    dn: cn=replica,cn=dc\3Dexample\2Cdc\3Dcom,cn=mapping tree,cn=config
    nsds5ReplicaPreciseTombstonePurging: on

When enabled, a range search is used to find the exact entries that need to
be removed:

    "(&(nsTombstoneCSN<=PURGE_CSN)(objectclass=nsTombstone))"

https://fedorahosted.org/389/ticket/47819

Jenkins: passed

Reviewed by: rmeggins(Thanks!)

Makefile.am

@@ -542,6 +542,7 @@ update_DATA = ldap/admin/src/scripts/exampleupdate.pl \
 	ldap/admin/src/scripts/50acctusabilityplugin.ldif \
 	ldap/admin/src/scripts/50automemberplugin.ldif \
 	ldap/admin/src/scripts/50memberofindex.ldif \
+	ldap/admin/src/scripts/50nstombstonecsn.ldif \
 	ldap/admin/src/scripts/50bitstringsyntaxplugin.ldif \
 	ldap/admin/src/scripts/50managedentriesplugin.ldif \
 	ldap/admin/src/scripts/50memberofplugin.ldif \

Makefile.in

@@ -1950,6 +1950,7 @@ update_DATA = ldap/admin/src/scripts/exampleupdate.pl \
 	ldap/admin/src/scripts/50acctusabilityplugin.ldif \
 	ldap/admin/src/scripts/50automemberplugin.ldif \
 	ldap/admin/src/scripts/50memberofindex.ldif \
+	ldap/admin/src/scripts/50nstombstonecsn.ldif \
 	ldap/admin/src/scripts/50bitstringsyntaxplugin.ldif \
 	ldap/admin/src/scripts/50managedentriesplugin.ldif \
 	ldap/admin/src/scripts/50memberofplugin.ldif \

aclocal.m4

@@ -220,6 +220,21 @@ m4_popdef([pkg_default])
 m4_popdef([pkg_description])
 ]) dnl PKG_NOARCH_INSTALLDIR
 
+
+# PKG_CHECK_VAR(VARIABLE, MODULE, CONFIG-VARIABLE,
+# [ACTION-IF-FOUND], [ACTION-IF-NOT-FOUND])
+# -------------------------------------------
+# Retrieves the value of the pkg-config variable for the given module.
+AC_DEFUN([PKG_CHECK_VAR],
+[AC_REQUIRE([PKG_PROG_PKG_CONFIG])dnl
+AC_ARG_VAR([$1], [value of $3 for $2, overriding pkg-config])dnl
+
+_PKG_CONFIG([$1], [variable="][$3]["], [$2])
+AS_VAR_COPY([$1], [pkg_cv_][$1])
+
+AS_VAR_IF([$1], [""], [$5], [$4])dnl
+])# PKG_CHECK_VAR
+
 # Copyright (C) 2002-2013 Free Software Foundation, Inc.
 #
 # This file is free software; the Free Software Foundation

config.guess

@@ -1,8 +1,10 @@
 #! /bin/sh
 # Attempt to guess a canonical system name.
-#   Copyright 1992-2013 Free Software Foundation, Inc.
+#   Copyright (C) 1992, 1993, 1994, 1995, 1996, 1997, 1998, 1999,
+#   2000, 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
+#   2011, 2012, 2013 Free Software Foundation, Inc.
 
-timestamp='2013-06-10'
+timestamp='2012-12-29'
 
 # This file is free software; you can redistribute it and/or modify it
 # under the terms of the GNU General Public License as published by
@@ -24,7 +26,7 @@ timestamp='2013-06-10'
 # program.  This Exception is an additional permission under section 7
 # of the GNU General Public License, version 3 ("GPLv3").
 #
-# Originally written by Per Bothner.
+# Originally written by Per Bothner. 
 #
 # You can get the latest version of this script from:
 # http://git.savannah.gnu.org/gitweb/?p=config.git;a=blob_plain;f=config.guess;hb=HEAD
@@ -50,7 +52,9 @@ version="\
 GNU config.guess ($timestamp)
 
 Originally written by Per Bothner.
-Copyright 1992-2013 Free Software Foundation, Inc.
+Copyright (C) 1992, 1993, 1994, 1995, 1996, 1997, 1998, 1999, 2000,
+2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011,
+2012, 2013 Free Software Foundation, Inc.
 
 This is free software; see the source for copying conditions.  There is NO
 warranty; not even for MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE."
@@ -132,27 +136,6 @@ UNAME_RELEASE=`(uname -r) 2>/dev/null` || UNAME_RELEASE=unknown
 UNAME_SYSTEM=`(uname -s) 2>/dev/null`  || UNAME_SYSTEM=unknown
 UNAME_VERSION=`(uname -v) 2>/dev/null` || UNAME_VERSION=unknown
 
-case "${UNAME_SYSTEM}" in
-Linux|GNU|GNU/*)
-	# If the system lacks a compiler, then just pick glibc.
-	# We could probably try harder.
-	LIBC=gnu
-
-	eval $set_cc_for_build
-	cat <<-EOF > $dummy.c
-	#include <features.h>
-	#if defined(__UCLIBC__)
-	LIBC=uclibc
-	#elif defined(__dietlibc__)
-	LIBC=dietlibc
-	#else
-	LIBC=gnu
-	#endif
-	EOF
-	eval `$CC_FOR_BUILD -E $dummy.c 2>/dev/null | grep '^LIBC'`
-	;;
-esac
-
 # Note: order is significant - the case branches are not exclusive.
 
 case "${UNAME_MACHINE}:${UNAME_SYSTEM}:${UNAME_RELEASE}:${UNAME_VERSION}" in
@@ -874,21 +857,21 @@ EOF
 	exit ;;
     *:GNU:*:*)
 	# the GNU system
-	echo `echo ${UNAME_MACHINE}|sed -e 's,[-/].*$,,'`-unknown-${LIBC}`echo ${UNAME_RELEASE}|sed -e 's,/.*$,,'`
+	echo `echo ${UNAME_MACHINE}|sed -e 's,[-/].*$,,'`-unknown-gnu`echo ${UNAME_RELEASE}|sed -e 's,/.*$,,'`
 	exit ;;
     *:GNU/*:*:*)
 	# other systems with GNU libc and userland
-	echo ${UNAME_MACHINE}-unknown-`echo ${UNAME_SYSTEM} | sed 's,^[^/]*/,,' | tr '[A-Z]' '[a-z]'``echo ${UNAME_RELEASE}|sed -e 's/[-(].*//'`-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-`echo ${UNAME_SYSTEM} | sed 's,^[^/]*/,,' | tr '[A-Z]' '[a-z]'``echo ${UNAME_RELEASE}|sed -e 's/[-(].*//'`-gnu
 	exit ;;
     i*86:Minix:*:*)
 	echo ${UNAME_MACHINE}-pc-minix
 	exit ;;
     aarch64:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     aarch64_be:Linux:*:*)
 	UNAME_MACHINE=aarch64_be
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     alpha:Linux:*:*)
 	case `sed -n '/^cpu model/s/^.*: \(.*\)/\1/p' < /proc/cpuinfo` in
@@ -901,54 +884,59 @@ EOF
 	  EV68*) UNAME_MACHINE=alphaev68 ;;
 	esac
 	objdump --private-headers /bin/sh | grep -q ld.so.1
-	if test "$?" = 0 ; then LIBC="gnulibc1" ; fi
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
-	exit ;;
-    arc:Linux:*:* | arceb:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	if test "$?" = 0 ; then LIBC="libc1" ; else LIBC="" ; fi
+	echo ${UNAME_MACHINE}-unknown-linux-gnu${LIBC}
 	exit ;;
     arm*:Linux:*:*)
 	eval $set_cc_for_build
 	if echo __ARM_EABI__ | $CC_FOR_BUILD -E - 2>/dev/null \
 	    | grep -q __ARM_EABI__
 	then
-	    echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	    echo ${UNAME_MACHINE}-unknown-linux-gnu
 	else
 	    if echo __ARM_PCS_VFP | $CC_FOR_BUILD -E - 2>/dev/null \
 		| grep -q __ARM_PCS_VFP
 	    then
-		echo ${UNAME_MACHINE}-unknown-linux-${LIBC}eabi
+		echo ${UNAME_MACHINE}-unknown-linux-gnueabi
 	    else
-		echo ${UNAME_MACHINE}-unknown-linux-${LIBC}eabihf
+		echo ${UNAME_MACHINE}-unknown-linux-gnueabihf
 	    fi
 	fi
 	exit ;;
     avr32*:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     cris:Linux:*:*)
-	echo ${UNAME_MACHINE}-axis-linux-${LIBC}
+	echo ${UNAME_MACHINE}-axis-linux-gnu
 	exit ;;
     crisv32:Linux:*:*)
-	echo ${UNAME_MACHINE}-axis-linux-${LIBC}
+	echo ${UNAME_MACHINE}-axis-linux-gnu
 	exit ;;
     frv:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     hexagon:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     i*86:Linux:*:*)
-	echo ${UNAME_MACHINE}-pc-linux-${LIBC}
+	LIBC=gnu
+	eval $set_cc_for_build
+	sed 's/^	//' << EOF >$dummy.c
+	#ifdef __dietlibc__
+	LIBC=dietlibc
+	#endif
+EOF
+	eval `$CC_FOR_BUILD -E $dummy.c 2>/dev/null | grep '^LIBC'`
+	echo "${UNAME_MACHINE}-pc-linux-${LIBC}"
 	exit ;;
     ia64:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     m32r*:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     m68*:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     mips:Linux:*:* | mips64:Linux:*:*)
 	eval $set_cc_for_build
@@ -967,63 +955,54 @@ EOF
 	#endif
 EOF
 	eval `$CC_FOR_BUILD -E $dummy.c 2>/dev/null | grep '^CPU'`
-	test x"${CPU}" != x && { echo "${CPU}-unknown-linux-${LIBC}"; exit; }
+	test x"${CPU}" != x && { echo "${CPU}-unknown-linux-gnu"; exit; }
 	;;
-    or1k:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
-	exit ;;
     or32:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     padre:Linux:*:*)
-	echo sparc-unknown-linux-${LIBC}
+	echo sparc-unknown-linux-gnu
 	exit ;;
     parisc64:Linux:*:* | hppa64:Linux:*:*)
-	echo hppa64-unknown-linux-${LIBC}
+	echo hppa64-unknown-linux-gnu
 	exit ;;
     parisc:Linux:*:* | hppa:Linux:*:*)
 	# Look for CPU level
 	case `grep '^cpu[^a-z]*:' /proc/cpuinfo 2>/dev/null | cut -d' ' -f2` in
-	  PA7*) echo hppa1.1-unknown-linux-${LIBC} ;;
-	  PA8*) echo hppa2.0-unknown-linux-${LIBC} ;;
-	  *)    echo hppa-unknown-linux-${LIBC} ;;
+	  PA7*) echo hppa1.1-unknown-linux-gnu ;;
+	  PA8*) echo hppa2.0-unknown-linux-gnu ;;
+	  *)    echo hppa-unknown-linux-gnu ;;
 	esac
 	exit ;;
     ppc64:Linux:*:*)
-	echo powerpc64-unknown-linux-${LIBC}
+	echo powerpc64-unknown-linux-gnu
 	exit ;;
     ppc:Linux:*:*)
-	echo powerpc-unknown-linux-${LIBC}
-	exit ;;
-    ppc64le:Linux:*:*)
-	echo powerpc64le-unknown-linux-${LIBC}
-	exit ;;
-    ppcle:Linux:*:*)
-	echo powerpcle-unknown-linux-${LIBC}
+	echo powerpc-unknown-linux-gnu
 	exit ;;
     s390:Linux:*:* | s390x:Linux:*:*)
-	echo ${UNAME_MACHINE}-ibm-linux-${LIBC}
+	echo ${UNAME_MACHINE}-ibm-linux
 	exit ;;
     sh64*:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     sh*:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     sparc:Linux:*:* | sparc64:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     tile*:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     vax:Linux:*:*)
-	echo ${UNAME_MACHINE}-dec-linux-${LIBC}
+	echo ${UNAME_MACHINE}-dec-linux-gnu
 	exit ;;
     x86_64:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     xtensa*:Linux:*:*)
-	echo ${UNAME_MACHINE}-unknown-linux-${LIBC}
+	echo ${UNAME_MACHINE}-unknown-linux-gnu
 	exit ;;
     i*86:DYNIX/ptx:4*:*)
 	# ptx 4.0 does uname -s correctly, with DYNIX/ptx in there.
@@ -1256,21 +1235,19 @@ EOF
 	exit ;;
     *:Darwin:*:*)
 	UNAME_PROCESSOR=`uname -p` || UNAME_PROCESSOR=unknown
-	eval $set_cc_for_build
-	if test "$UNAME_PROCESSOR" = unknown ; then
-	    UNAME_PROCESSOR=powerpc
-	fi
-	if [ "$CC_FOR_BUILD" != 'no_compiler_found' ]; then
-	    if (echo '#ifdef __LP64__'; echo IS_64BIT_ARCH; echo '#endif') | \
-		(CCOPTS= $CC_FOR_BUILD -E - 2>/dev/null) | \
-		grep IS_64BIT_ARCH >/dev/null
-	    then
-		case $UNAME_PROCESSOR in
-		    i386) UNAME_PROCESSOR=x86_64 ;;
-		    powerpc) UNAME_PROCESSOR=powerpc64 ;;
-		esac
-	    fi
-	fi
+	case $UNAME_PROCESSOR in
+	    i386)
+		eval $set_cc_for_build
+		if [ "$CC_FOR_BUILD" != 'no_compiler_found' ]; then
+		  if (echo '#ifdef __LP64__'; echo IS_64BIT_ARCH; echo '#endif') | \
+		      (CCOPTS= $CC_FOR_BUILD -E - 2>/dev/null) | \
+		      grep IS_64BIT_ARCH >/dev/null
+		  then
+		      UNAME_PROCESSOR="x86_64"
+		  fi
+		fi ;;
+	    unknown) UNAME_PROCESSOR=powerpc ;;
+	esac
 	echo ${UNAME_PROCESSOR}-apple-darwin${UNAME_RELEASE}
 	exit ;;
     *:procnto*:*:* | *:QNX:[0123456789]*:*)

config.sub

@@ -1,8 +1,10 @@
 #! /bin/sh
 # Configuration validation subroutine script.
-#   Copyright 1992-2013 Free Software Foundation, Inc.
+#   Copyright (C) 1992, 1993, 1994, 1995, 1996, 1997, 1998, 1999,
+#   2000, 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010,
+#   2011, 2012, 2013 Free Software Foundation, Inc.
 
-timestamp='2013-04-24'
+timestamp='2012-12-29'
 
 # This file is free software; you can redistribute it and/or modify it
 # under the terms of the GNU General Public License as published by
@@ -68,7 +70,9 @@ Report bugs and patches to <[email protected]>."
 version="\
 GNU config.sub ($timestamp)
 
-Copyright 1992-2013 Free Software Foundation, Inc.
+Copyright (C) 1992, 1993, 1994, 1995, 1996, 1997, 1998, 1999, 2000,
+2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011,
+2012, 2013 Free Software Foundation, Inc.
 
 This is free software; see the source for copying conditions.  There is NO
 warranty; not even for MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE."
@@ -252,7 +256,7 @@ case $basic_machine in
 	| alpha | alphaev[4-8] | alphaev56 | alphaev6[78] | alphapca5[67] \
 	| alpha64 | alpha64ev[4-8] | alpha64ev56 | alpha64ev6[78] | alpha64pca5[67] \
 	| am33_2.0 \
-	| arc | arceb \
+	| arc \
 	| arm | arm[bl]e | arme[lb] | armv[2-8] | armv[3-8][lb] | armv7[arm] \
 	| avr | avr32 \
 	| be32 | be64 \
@@ -286,17 +290,16 @@ case $basic_machine in
 	| mipsisa64r2 | mipsisa64r2el \
 	| mipsisa64sb1 | mipsisa64sb1el \
 	| mipsisa64sr71k | mipsisa64sr71kel \
-	| mipsr5900 | mipsr5900el \
 	| mipstx39 | mipstx39el \
 	| mn10200 | mn10300 \
 	| moxie \
 	| mt \
 	| msp430 \
 	| nds32 | nds32le | nds32be \
-	| nios | nios2 | nios2eb | nios2el \
+	| nios | nios2 \
 	| ns16k | ns32k \
 	| open8 \
-	| or1k | or32 \
+	| or32 \
 	| pdp10 | pdp11 | pj | pjl \
 	| powerpc | powerpc64 | powerpc64le | powerpcle \
 	| pyramid \
@@ -366,7 +369,7 @@ case $basic_machine in
 	| aarch64-* | aarch64_be-* \
 	| alpha-* | alphaev[4-8]-* | alphaev56-* | alphaev6[78]-* \
 	| alpha64-* | alpha64ev[4-8]-* | alpha64ev56-* | alpha64ev6[78]-* \
-	| alphapca5[67]-* | alpha64pca5[67]-* | arc-* | arceb-* \
+	| alphapca5[67]-* | alpha64pca5[67]-* | arc-* \
 	| arm-*  | armbe-* | armle-* | armeb-* | armv*-* \
 	| avr-* | avr32-* \
 	| be32-* | be64-* \
@@ -404,13 +407,12 @@ case $basic_machine in
 	| mipsisa64r2-* | mipsisa64r2el-* \
 	| mipsisa64sb1-* | mipsisa64sb1el-* \
 	| mipsisa64sr71k-* | mipsisa64sr71kel-* \
-	| mipsr5900-* | mipsr5900el-* \
 	| mipstx39-* | mipstx39el-* \
 	| mmix-* \
 	| mt-* \
 	| msp430-* \
 	| nds32-* | nds32le-* | nds32be-* \
-	| nios-* | nios2-* | nios2eb-* | nios2el-* \
+	| nios-* | nios2-* \
 	| none-* | np1-* | ns16k-* | ns32k-* \
 	| open8-* \
 	| orion-* \
@@ -1352,7 +1354,7 @@ case $os in
 	-gnu* | -bsd* | -mach* | -minix* | -genix* | -ultrix* | -irix* \
 	      | -*vms* | -sco* | -esix* | -isc* | -aix* | -cnk* | -sunos | -sunos[34]*\
 	      | -hpux* | -unos* | -osf* | -luna* | -dgux* | -auroraux* | -solaris* \
-	      | -sym* | -kopensolaris* | -plan9* \
+	      | -sym* | -kopensolaris* \
 	      | -amigaos* | -amigados* | -msdos* | -newsos* | -unicos* | -aof* \
 	      | -aos* | -aros* \
 	      | -nindy* | -vxsim* | -vxworks* | -ebmon* | -hms* | -mvs* \
@@ -1498,6 +1500,9 @@ case $os in
 	-aros*)
 		os=-aros
 		;;
+	-kaos*)
+		os=-kaos
+		;;
 	-zvmoe)
 		os=-zvmoe
 		;;
@@ -1589,9 +1594,6 @@ case $basic_machine in
 	mips*-*)
 		os=-elf
 		;;
-	or1k-*)
-		os=-elf
-		;;
 	or32-*)
 		os=-coff
 		;;

dirsrvtests/data/README

@@ -0,0 +1,11 @@
+DATA DIRECTORY README
+
+This directory is used for storing LDIF files used by the dirsrvtests scripts. 
+This directory can be retrieved via getDir() from the DirSrv class.
+
+Example:
+
+    data_dir_path = topology.standalone.getDir(__file__, DATA_DIR)
+
+    ldif_file = data_dir_path + "ticket44444/1000entries.ldif"
+

dirsrvtests/tickets/ticket47819_test.py

@@ -0,0 +1,345 @@
+import os
+import sys
+import time
+import ldap
+import logging
+import socket
+import pytest
+from lib389 import DirSrv, Entry, tools, tasks
+from lib389.tools import DirSrvTools
+from lib389._constants import *
+from lib389.properties import *
+from lib389.tasks import *
+from constants import *
+
+log = logging.getLogger(__name__)
+
+installation_prefix = None
+
+
+class TopologyStandalone(object):
+    def __init__(self, standalone):
+        standalone.open()
+        self.standalone = standalone
+
+
[email protected](scope="module")
+def topology(request):
+    '''
+        This fixture is used to standalone topology for the 'module'.
+        At the beginning, It may exists a standalone instance.
+        It may also exists a backup for the standalone instance.
+
+        Principle:
+            If standalone instance exists:
+                restart it
+            If backup of standalone exists:
+                create/rebind to standalone
+
+                restore standalone instance from backup
+            else:
+                Cleanup everything
+                    remove instance
+                    remove backup
+                Create instance
+                Create backup
+    '''
+    global installation_prefix
+
+    if installation_prefix:
+        args_instance[SER_DEPLOYED_DIR] = installation_prefix
+
+    standalone = DirSrv(verbose=False)
+
+    # Args for the standalone instance
+    args_instance[SER_HOST] = HOST_STANDALONE
+    args_instance[SER_PORT] = PORT_STANDALONE
+    args_instance[SER_SERVERID_PROP] = SERVERID_STANDALONE
+    args_standalone = args_instance.copy()
+    standalone.allocate(args_standalone)
+
+    # Get the status of the backups
+    backup_standalone = standalone.checkBackupFS()
+
+    # Get the status of the instance and restart it if it exists
+    instance_standalone = standalone.exists()
+    if instance_standalone:
+        # assuming the instance is already stopped, just wait 5 sec max
+        standalone.stop(timeout=5)
+        standalone.start(timeout=60)
+
+    if backup_standalone:
+        # The backup exist, assuming it is correct
+        # we just re-init the instance with it
+        if not instance_standalone:
+            standalone.create()
+            # Used to retrieve configuration information (dbdir, confdir...)
+            standalone.open()
+
+        # restore standalone instance from backup
+        standalone.stop(timeout=10)
+        standalone.restoreFS(backup_standalone)
+        standalone.start(timeout=60)
+
+    else:
+        # We should be here only in two conditions
+        #      - This is the first time a test involve standalone instance
+        #      - Something weird happened (instance/backup destroyed)
+        #        so we discard everything and recreate all
+
+        # Remove the backup. So even if we have a specific backup file
+        # (e.g backup_standalone) we clear backup that an instance may have created
+        if backup_standalone:
+            standalone.clearBackupFS()
+
+        # Remove the instance
+        if instance_standalone:
+            standalone.delete()
+
+        # Create the instance
+        standalone.create()
+
+        # Used to retrieve configuration information (dbdir, confdir...)
+        standalone.open()
+
+        # Time to create the backups
+        standalone.stop(timeout=10)
+        standalone.backupfile = standalone.backupFS()
+        standalone.start(timeout=60)
+
+    # Here we have standalone instance up and running
+    # Either coming from a backup recovery
+    # or from a fresh (re)init
+    # Time to return the topology
+    return TopologyStandalone(standalone)
+
+
+def test_ticket47819(topology):
+    """
+        Testing precise tombstone purging:
+            [1]  Make sure "nsTombstoneCSN" is added to new tombstones
+            [2]  Make sure an import of a replication ldif adds "nsTombstoneCSN"
+                 to old tombstones
+            [4]  Test fixup task
+            [3]  Make sure tombstone purging works
+    """
+
+    log.info('Testing Ticket 47819 - Test precise tombstone purging')
+
+    #
+    # Setup Replication
+    #
+    log.info('Setting up replication...')
+    topology.standalone.replica.enableReplication(suffix=DEFAULT_SUFFIX, role=REPLICAROLE_MASTER,
+                                                  replicaId=REPLICAID_MASTER_1)
+
+    #
+    # Part 1 create a tombstone entry and make sure nsTombstoneCSN is added
+    #
+    log.info('Part 1:  Add and then delete an entry to create a tombstone...')
+
+    try:
+        topology.standalone.add_s(Entry(('cn=entry1,dc=example,dc=com', {
+                                  'objectclass': 'top person'.split(),
+                                  'sn': 'user',
+                                  'cn': 'entry1'})))
+    except ldap.LDAPError, e:
+        log.error('Failed to add entry: ' + e.message['desc'])
+        assert False
+
+    try:
+        topology.standalone.delete_s('cn=entry1,dc=example,dc=com')
+    except ldap.LDAPError, e:
+        log.error('Failed to delete entry: ' + e.message['desc'])
+        assert False
+
+    log.info('Search for tombstone entries...')
+    try:
+        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
+                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
+        if not entries:
+            log.fatal('Search failed to the new tombstone(nsTombstoneCSN is probably missing).')
+            assert False
+    except ldap.LDAPError, e:
+        log.fatal('Search failed: ' + e.message['desc'])
+        assert False
+
+    log.info('Part 1 - passed')
+
+    #
+    # Part 2 - import ldif with tombstones missing 'nsTombstoneCSN'
+    #
+    # First, export the replication ldif, edit the file(remove nstombstonecsn),
+    # and reimport it.
+    #
+    log.info('Part 2:  Exporting replication ldif...')
+
+    # Get the the full path and name for our LDIF we will be exporting
+    ldif_file = topology.standalone.getDir(__file__, TMP_DIR) + "export.ldif"
+
+    args = {EXPORT_REPL_INFO: True,
+            TASK_WAIT: True}
+    exportTask = Tasks(topology.standalone)
+    try:
+        exportTask.exportLDIF(DEFAULT_SUFFIX, None, ldif_file, args)
+    except ValueError:
+        assert False
+
+    # open the ldif file, get the lines, then rewrite the file
+    ldif = open(ldif_file, "r")
+    lines = ldif.readlines()
+    ldif.close()
+
+    ldif = open(ldif_file, "w")
+    for line in lines:
+        if not line.lower().startswith('nstombstonecsn'):
+            ldif.write(line)
+    ldif.close()
+
+    # import the new ldif file
+    log.info('Import replication LDIF file...')
+    importTask = Tasks(topology.standalone)
+    args = {TASK_WAIT: True}
+    try:
+        importTask.importLDIF(DEFAULT_SUFFIX, None, ldif_file, args)
+        os.remove(ldif_file)
+    except ValueError:
+        os.remove(ldif_file)
+        assert False
+
+    # Search for the tombstone again
+    log.info('Search for tombstone entries...')
+    try:
+        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
+                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
+        if not entries:
+            log.fatal('Search failed to fine the new tombstone(nsTombstoneCSN is probably missing).')
+            assert False
+    except ldap.LDAPError, e:
+        log.fatal('Search failed: ' + e.message['desc'])
+        assert False
+
+    log.info('Part 2 - passed')
+
+    #
+    # Part 3 - test fixup task
+    #
+    log.info('Part 4:  test the fixup task')
+
+    # Run fixup task using the strip option.  This removes nsTombstoneCSN
+    # so we can test if the fixup task works.
+    args = {TASK_WAIT: True,
+            TASK_TOMB_STRIP: True}
+    fixupTombTask = Tasks(topology.standalone)
+    try:
+        fixupTombTask.fixupTombstones(DEFAULT_BENAME, args)
+    except:
+        assert False
+
+    # Search for tombstones with nsTombstoneCSN - better not find any
+    log.info('Search for tombstone entries...')
+    try:
+        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
+                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
+        if entries:
+            log.fatal('Search found tombstones with nsTombstoneCSN')
+            assert False
+    except ldap.LDAPError, e:
+        log.fatal('Search failed: ' + e.message['desc'])
+        assert False
+
+    # Now run the fixup task
+    args = {TASK_WAIT: True}
+    fixupTombTask = Tasks(topology.standalone)
+    try:
+        fixupTombTask.fixupTombstones(DEFAULT_BENAME, args)
+    except:
+        assert False
+
+    # Search for tombstones with nsTombstoneCSN - better find some
+    log.info('Search for tombstone entries...')
+    try:
+        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
+                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
+        if not entries:
+            log.fatal('Search did not find any fixed-up tombstones')
+            assert False
+    except ldap.LDAPError, e:
+        log.fatal('Search failed: ' + e.message['desc'])
+        assert False
+
+    log.info('Part 3 - passed')
+
+    #
+    # Part 4 - Test tombstone purging
+    #
+    log.info('Part 4:  test tombstone purging...')
+
+    args = {REPLICA_PRECISE_PURGING: 'on',
+            REPLICA_PURGE_DELAY: '5',
+            REPLICA_PURGE_INTERVAL: '5'}
+    try:
+        topology.standalone.replica.setProperties(DEFAULT_SUFFIX, None, None, args)
+    except:
+        log.fatal('Failed to configure replica')
+        assert False
+
+    # Wait for the interval to pass
+    log.info('Wait for tombstone purge interval to pass...')
+    time.sleep(6)
+
+    # Add an entry to trigger replication
+    log.info('Perform an update to help trigger tombstone purging...')
+    try:
+        topology.standalone.add_s(Entry(('cn=test_entry,dc=example,dc=com', {
+                                  'objectclass': 'top person'.split(),
+                                  'sn': 'user',
+                                  'cn': 'entry1'})))
+    except ldap.LDAPError, e:
+        log.error('Failed to add entry: ' + e.message['desc'])
+        assert False
+
+    # Wait for the interval to pass again
+    log.info('Wait for tombstone purge interval to pass again...')
+    time.sleep(10)
+
+    # search for tombstones, there should be none
+    log.info('Search for tombstone entries...')
+    try:
+        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
+                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
+        if entries:
+            log.fatal('Search unexpectedly found tombstones')
+            assert False
+    except ldap.LDAPError, e:
+        log.fatal('Search failed: ' + e.message['desc'])
+        assert False
+
+    log.info('Part 4 - passed')
+
+    #
+    # If we got here we passed!
+    #
+    log.info('Ticket47819 Test - Passed')
+
+
+def test_ticket47819_final(topology):
+    topology.standalone.stop(timeout=10)
+
+
+def run_isolated():
+    '''
+        run_isolated is used to run these test cases independently of a test scheduler (xunit, py.test..)
+        To run isolated without py.test, you need to
+            - edit this file and comment '@pytest.fixture' line before 'topology' function.
+            - set the installation prefix
+            - run this program
+    '''
+    global installation_prefix
+    installation_prefix = None
+
+    topo = topology(True)
+    test_ticket47819(topo)
+
+if __name__ == '__main__':
+    run_isolated()

dirsrvtests/tmp/README

@@ -0,0 +1,10 @@
+TMP DIRECTORY README
+
+This directory is used to store files(LDIFs, etc) that are created during the ticket script runtime.  The script is also responsible for removing any files it places in this directory.  This directory can be retrieved via getDir() from the DirSrv class.
+
+Example:
+
+    tmp_dir_path = topology.standalone.getDir(__file__, TMP_DIR)
+
+    new_ldif = tmp_dir_path + "export.ldif"
+

ldap/admin/src/scripts/50nstombstonecsn.ldif

@@ -0,0 +1,7 @@
+dn: cn=nsTombstoneCSN,cn=default indexes,cn=config,cn=ldbm database,cn=plugins,cn=config
+changetype: add
+objectclass: top
+objectclass: nsIndex
+cn: nsTombstoneCSN
+nssystemindex: true
+nsindextype: eq

ldap/ldif/template-dse.ldif.in

@@ -967,6 +967,13 @@ cn: uniquemember
 nssystemindex: false
 nsindextype: eq
 
+dn: cn=nsTombstoneCSN,cn=default indexes, cn=config,cn=ldbm database,cn=plugins,cn=config
+objectclass: top
+objectclass: nsIndex
+cn: nsTombstoneCSN
+nssystemindex: true
+nsindextype: eq
+
 dn: cn=monitor, cn=ldbm database, cn=plugins, cn=config
 objectclass: top
 objectclass: extensibleObject

ldap/schema/01core389.ldif

@@ -299,6 +299,8 @@ attributeTypes: ( 2.16.840.1.113730.3.1.2304 NAME 'nsslapd-dynamic-plugins' DESC
 attributeTypes: ( 2.16.840.1.113730.3.1.2305 NAME 'nsslapd-moddn-aci' DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 SINGLE-VALUE X-ORIGIN 'Netscape Directory Server' )
 attributeTypes: ( 2.16.840.1.113730.3.1.2306 NAME 'nsslapd-return-default-opattr' DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 USAGE directoryOperation X-ORIGIN 'Netscape Directory Server' )
 attributeTypes: ( 2.16.840.1.113730.3.1.2307 NAME 'nsslapd-allow-hashed-passwords' DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 SINGLE-VALUE X-ORIGIN 'Netscape Directory Server' )
+attributeTypes: ( 2.16.840.1.113730.3.1.2308 NAME 'nstombstonecsn' DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 SINGLE-VALUE X-ORIGIN 'Netscape Directory Server' )
+attributeTypes: ( 2.16.840.1.113730.3.1.2309 NAME 'nsds5ReplicaPreciseTombstonePurging' DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 SINGLE-VALUE X-ORIGIN 'Netscape Directory Server' )
 #
 # objectclasses
 #
@@ -308,8 +310,8 @@ objectClasses: ( 2.16.840.1.113730.3.2.44 NAME 'nsIndex' DESC 'Netscape defined
 objectClasses: ( 2.16.840.1.113730.3.2.109 NAME 'nsBackendInstance' DESC 'Netscape defined objectclass' SUP top  MUST ( CN ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.110 NAME 'nsMappingTree' DESC 'Netscape defined objectclass' SUP top  MUST ( CN ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.104 NAME 'nsContainer' DESC 'Netscape defined objectclass' SUP top  MUST ( CN ) X-ORIGIN 'Netscape Directory Server' )
-objectClasses: ( 2.16.840.1.113730.3.2.108 NAME 'nsDS5Replica' DESC 'Netscape defined objectclass' SUP top  MUST ( nsDS5ReplicaRoot $  nsDS5ReplicaId ) MAY (cn $ nsds5ReplicaCleanRUV $ nsds5ReplicaAbortCleanRUV $ nsDS5ReplicaType $ nsDS5ReplicaBindDN $ nsState $ nsDS5ReplicaName $ nsDS5Flags $ nsDS5Task $ nsDS5ReplicaReferral $ nsDS5ReplicaAutoReferral $ nsds5ReplicaPurgeDelay $ nsds5ReplicaTombstonePurgeInterval $ nsds5ReplicaChangeCount $ nsds5ReplicaLegacyConsumer $ nsds5ReplicaProtocolTimeout $ nsds5ReplicaBackoffMin $ nsds5ReplicaBackoffMax ) X-ORIGIN 'Netscape Directory Server' )
-objectClasses: ( 2.16.840.1.113730.3.2.113 NAME 'nsTombstone' DESC 'Netscape defined objectclass' SUP top MAY ( nsParentUniqueId $ nscpEntryDN ) X-ORIGIN 'Netscape Directory Server' )
+objectClasses: ( 2.16.840.1.113730.3.2.108 NAME 'nsDS5Replica' DESC 'Netscape defined objectclass' SUP top  MUST ( nsDS5ReplicaRoot $  nsDS5ReplicaId ) MAY (cn $ nsds5ReplicaPreciseTombstonePurging $ nsds5ReplicaCleanRUV $ nsds5ReplicaAbortCleanRUV $ nsDS5ReplicaType $ nsDS5ReplicaBindDN $ nsState $ nsDS5ReplicaName $ nsDS5Flags $ nsDS5Task $ nsDS5ReplicaReferral $ nsDS5ReplicaAutoReferral $ nsds5ReplicaPurgeDelay $ nsds5ReplicaTombstonePurgeInterval $ nsds5ReplicaChangeCount $ nsds5ReplicaLegacyConsumer $ nsds5ReplicaProtocolTimeout $ nsds5ReplicaBackoffMin $ nsds5ReplicaBackoffMax ) X-ORIGIN 'Netscape Directory Server' )
+objectClasses: ( 2.16.840.1.113730.3.2.113 NAME 'nsTombstone' DESC 'Netscape defined objectclass' SUP top MAY ( nstombstonecsn $ nsParentUniqueId $ nscpEntryDN ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.103 NAME 'nsDS5ReplicationAgreement' DESC 'Netscape defined objectclass' SUP top MUST ( cn ) MAY ( nsds5ReplicaCleanRUVNotified $ nsDS5ReplicaHost $ nsDS5ReplicaPort $ nsDS5ReplicaTransportInfo $ nsDS5ReplicaBindDN $ nsDS5ReplicaCredentials $ nsDS5ReplicaBindMethod $ nsDS5ReplicaRoot $ nsDS5ReplicatedAttributeList $ nsDS5ReplicatedAttributeListTotal $ nsDS5ReplicaUpdateSchedule $ nsds5BeginReplicaRefresh $ description $ nsds50ruv $ nsruvReplicaLastModified $ nsds5ReplicaTimeout $ nsds5replicaChangesSentSinceStartup $ nsds5replicaLastUpdateEnd $ nsds5replicaLastUpdateStart $ nsds5replicaLastUpdateStatus $ nsds5replicaUpdateInProgress $ nsds5replicaLastInitEnd $ nsds5ReplicaEnabled $ nsds5replicaLastInitStart $ nsds5replicaLastInitStatus $ nsds5debugreplicatimeout $ nsds5replicaBusyWaitTime $ nsds5ReplicaStripAttrs $ nsds5replicaSessionPauseTime $ nsds5ReplicaProtocolTimeout ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.39 NAME 'nsslapdConfig' DESC 'Netscape defined objectclass' SUP top MAY ( cn ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.317 NAME 'nsSaslMapping' DESC 'Netscape defined objectclass' SUP top MUST ( cn $ nsSaslMapRegexString $ nsSaslMapBaseDNTemplate $ nsSaslMapFilterTemplate ) MAY ( nsSaslMapPriority ) X-ORIGIN 'Netscape Directory Server' )

ldap/servers/plugins/replication/repl5.h

@@ -173,6 +173,7 @@ extern const char *type_nsds5ReplicaStripAttrs;
 extern const char *type_replicaProtocolTimeout;
 extern const char *type_replicaBackoffMin;
 extern const char *type_replicaBackoffMax;
+extern const char *type_replicaPrecisePurge;
 
 /* Attribute names for windows replication agreements */
 extern const char *type_nsds7WindowsReplicaArea;
@@ -595,7 +596,6 @@ void replica_destroy_dn_hash ();
 int replica_add_by_dn (const char *dn);
 int replica_delete_by_dn (const char *dn);
 int replica_is_being_configured (const char *dn);
-const CSN * _get_deletion_csn(Slapi_Entry *e);
 int legacy_consumer_init_referrals (Replica *r);
 void consumer5_set_mapping_tree_state_for_replica(const Replica *r, RUV *supplierRuv);
 Object *replica_get_for_backend (const char *be_name);
@@ -619,6 +619,8 @@ void replica_set_backoff_max(Replica *r, PRUint64 max);
 int replica_get_agmt_count(Replica *r);
 void replica_incr_agmt_count(Replica *r);
 void replica_decr_agmt_count(Replica *r);
+PRUint64 replica_get_precise_purging(Replica *r);
+void replica_set_precise_purging(Replica *r, PRUint64 on_off);
 
 /* The functions below handles the state flag */
 /* Current internal state flags */

ldap/servers/plugins/replication/repl5_replica.c

@@ -95,6 +95,7 @@ struct replica {
 	Slapi_Counter *protocol_timeout;/* protocol shutdown timeout */
 	Slapi_Counter *backoff_min;		/* backoff retry minimum */
 	Slapi_Counter *backoff_max;		/* backoff retry maximum */
+	Slapi_Counter *precise_purging;		/* Enable precise tombstone purging */
 	PRUint64 agmt_count;			/* Number of agmts */
 };
 
@@ -1086,7 +1087,7 @@ char *replica_get_generation (const Replica *r)
     if (r && r->repl_ruv)
     {        
         replica_lock(r->repl_lock);
-	        
+
         if (rc == 0)
             gen = ruv_get_replica_generation ((RUV*)object_get_data (r->repl_ruv));
 
@@ -1789,6 +1790,7 @@ _replica_init_from_config (Replica *r, Slapi_Entry *e, char *errortext)
     Slapi_Attr *a = NULL;
     Slapi_Attr *attr;
     CSNGen *gen;
+    char *precise_purging = NULL;
     char buf [SLAPI_DSE_RETURNTEXT_SIZE];
     char *errormsg = errortext? errortext : buf;
     char *val;
@@ -1874,6 +1876,25 @@ _replica_init_from_config (Replica *r, Slapi_Entry *e, char *errortext)
         slapi_counter_set_value(r->protocol_timeout, ptimeout);
     }
 
+    /* check for precise tombstone purging */
+    precise_purging = slapi_entry_attr_get_charptr(e, type_replicaPrecisePurge);
+    if(precise_purging){
+        if (strcasecmp(precise_purging, "on") == 0){
+            slapi_counter_set_value(r->precise_purging, 1);
+        } else if (strcasecmp(precise_purging, "off") == 0){
+            slapi_counter_set_value(r->precise_purging, 0);
+        } else{
+            /* Invalid value */
+            slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+                    "Invalid value for %s: %s  Using default value (off)\n",
+                    type_replicaPrecisePurge, precise_purging);
+            slapi_counter_set_value(r->precise_purging, 0);
+        }
+        slapi_ch_free_string(&precise_purging);
+    } else {
+        slapi_counter_set_value(r->precise_purging, 0);
+    }
+
     /* get replica flags */
     r->repl_flags = slapi_entry_attr_get_ulong(e, attr_flags);
 
@@ -2862,7 +2883,7 @@ replica_ruv_smods_for_op( Slapi_PBlock *pb, char **uniqueid, Slapi_Mods **smods
 
 
 const CSN *
-_get_deletion_csn(Slapi_Entry *e)
+entry_get_deletion_csn(Slapi_Entry *e)
 {
 	const CSN *deletion_csn = NULL;
 
@@ -2950,7 +2971,7 @@ process_reap_entry (Slapi_Entry *entry, void *cb_data)
 	   objectclass attribute values - if we need more attributes returned by the
 	   search in the future, see _replica_reap_tombstones below and add more to the
 	   attrs array */
-	deletion_csn = _get_deletion_csn(entry);
+	deletion_csn = entry_get_deletion_csn(entry);
 
 	if ((NULL == deletion_csn || csn_compare(deletion_csn, purge_csn) < 0) &&
 		(!is_ruv_tombstone_entry(entry))) {
@@ -3043,9 +3064,26 @@ _replica_reap_tombstones(void *arg)
 	if (NULL != purge_csn) 
 	{
 		LDAPControl **ctrls;
-		int oprc;
 		reap_callback_data cb_data;
+		char deletion_csn_str[CSN_STRSIZE];
+		char tombstone_filter[128];
 		char **attrs = NULL;
+		int oprc;
+
+		if (replica_get_precise_purging(replica)){
+			/*
+			 * Using precise tombstone purging.  Create filter to lookup the exact
+			 * entries that need to be purged by using a range search on the new
+			 * tombstone csn index.
+			 */
+			csn_as_string(purge_csn, PR_FALSE, deletion_csn_str);
+			PR_snprintf(tombstone_filter, 128,
+			        "(&(%s<=%s)(objectclass=nsTombstone))", SLAPI_ATTR_TOMBSTONE_CSN,
+			        csn_as_string(purge_csn, PR_FALSE, deletion_csn_str));
+		} else {
+			/* Use the old inefficient filter */
+			PR_snprintf(tombstone_filter, 128, "(objectclass=nsTombstone)");
+		}
 
 		/* we just need the objectclass - for the deletion csn
 		   and the dn and nsuniqueid - for possible deletion
@@ -3055,6 +3093,7 @@ _replica_reap_tombstones(void *arg)
 		charray_add(&attrs, slapi_ch_strdup("objectclass"));
 		charray_add(&attrs, slapi_ch_strdup("nsuniqueid"));
 		charray_add(&attrs, slapi_ch_strdup("tombstonenumsubordinates"));
+		charray_add(&attrs, slapi_ch_strdup(SLAPI_ATTR_TOMBSTONE_CSN));
 
 		ctrls = (LDAPControl **)slapi_ch_calloc (3, sizeof (LDAPControl *));
 		ctrls[0] = create_managedsait_control();
@@ -3062,7 +3101,7 @@ _replica_reap_tombstones(void *arg)
 		ctrls[2] = NULL;
 		pb = slapi_pblock_new();
 		slapi_search_internal_set_pb(pb, slapi_sdn_get_dn(replica->repl_root),
-									 LDAP_SCOPE_SUBTREE, "(objectclass=nstombstone)",
+									 LDAP_SCOPE_SUBTREE, tombstone_filter,
 									 attrs, 0, ctrls, NULL,
 									 repl_get_plugin_identity(PLUGIN_MULTIMASTER_REPLICATION),
 									 OP_FLAG_REVERSE_CANDIDATE_ORDER);
@@ -4067,6 +4106,24 @@ replica_set_backoff_max(Replica *r, PRUint64 max)
 	}
 }
 
+void
+replica_set_precise_purging(Replica *r, PRUint64 on_off)
+{
+	if(r){
+		slapi_counter_set_value(r->precise_purging, on_off);
+	}
+}
+
+PRUint64
+replica_get_precise_purging(Replica *r)
+{
+	if(r){
+		return slapi_counter_get_value(r->precise_purging);
+	} else {
+		return 0;
+	}
+}
+
 int
 replica_get_agmt_count(Replica *r)
 {

ldap/servers/plugins/replication/repl5_replica_config.c

@@ -381,24 +381,24 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
                  */
                 if (strcasecmp (config_attr, attr_replicaBindDn) == 0)
                 {
-			*returncode = replica_config_change_updatedn (r, mods[i], errortext, apply_mods);
+                    *returncode = replica_config_change_updatedn (r, mods[i], errortext, apply_mods);
                 }
                 else if (strcasecmp (config_attr, attr_replicaBindDnGroup) == 0)
                 {
-			*returncode = replica_config_change_updatedngroup (r, mods[i], errortext, apply_mods);
+                    *returncode = replica_config_change_updatedngroup (r, mods[i], errortext, apply_mods);
                 }
                 else if (strcasecmp (config_attr, attr_replicaBindDnGroupCheckInterval) == 0)
                 {
-			replica_set_groupdn_checkinterval (r, -1);
-		}
+                    replica_set_groupdn_checkinterval (r, -1);
+                }
                 else if (strcasecmp (config_attr, attr_replicaReferral) == 0)
                 {
                     if (apply_mods) {
-				        replica_set_referrals(r, NULL);
-						if (!replica_is_legacy_consumer (r)) {
-							consumer5_set_mapping_tree_state_for_replica(r, NULL);
-						}
-					}
+                        replica_set_referrals(r, NULL);
+                        if (!replica_is_legacy_consumer (r)) {
+                             consumer5_set_mapping_tree_state_for_replica(r, NULL);
+                        }
+                    }
                 }
                 else if (strcasecmp (config_attr, type_replicaLegacyConsumer) == 0)
                 {
@@ -428,6 +428,11 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
                     if (apply_mods)
                         replica_set_backoff_max(r, PROTOCOL_BACKOFF_MAXIMUM);
                 }
+                else if (strcasecmp (config_attr, type_replicaPrecisePurge) == 0 )
+                {
+                    if (apply_mods)
+                        replica_set_precise_purging(r, 0);
+                    }
                 else
                 {
                     *returncode = LDAP_UNWILLING_TO_PERFORM;
@@ -448,16 +453,16 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
                 }
 
                 if (strcasecmp (config_attr, attr_replicaBindDn) == 0) {
-			*returncode = replica_config_change_updatedn (r, mods[i], errortext, apply_mods);
+                    *returncode = replica_config_change_updatedn (r, mods[i], errortext, apply_mods);
                 }
                 else if (strcasecmp (config_attr, attr_replicaBindDnGroup) == 0)
                 {
-			*returncode = replica_config_change_updatedngroup (r, mods[i], errortext, apply_mods);
+                    *returncode = replica_config_change_updatedngroup (r, mods[i], errortext, apply_mods);
                 }
                 else if (strcasecmp (config_attr, attr_replicaBindDnGroupCheckInterval) == 0)
                 {
-			int interval = atoi(config_attr_value);
-			replica_set_groupdn_checkinterval (r, interval);
+                    int interval = atoi(config_attr_value);
+                    replica_set_groupdn_checkinterval (r, interval);
                 }
                 else if (strcasecmp (config_attr, attr_replicaType) == 0)
 			    {
@@ -586,6 +591,35 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
                         replica_set_backoff_max(r, val);
                     }
                 }
+                else if (strcasecmp (config_attr, type_replicaPrecisePurge) == 0 )
+                {
+                    if (apply_mods)
+                    {
+                        if (apply_mods && config_attr_value[0])
+                        {
+                            PRUint64 on_off = 0;
+
+                            if (strcasecmp(config_attr_value, "on") == 0){
+                                on_off = 1;
+                            } else if (strcasecmp(config_attr_value, "off") == 0){
+                                on_off = 0;
+                            } else{
+                                /* Invalid value */
+                                *returncode = LDAP_UNWILLING_TO_PERFORM;
+                                PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE,
+                                        "Invalid value for %s: %s  Value should be \"on\" or \"off\"\n",
+                                        type_replicaPrecisePurge, config_attr_value);
+                                slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+                                        "Invalid value for %s: %s  Value should be \"on\" or \"off\")\n",
+                                        type_replicaPrecisePurge, config_attr_value);
+                                break;
+                            }
+                            replica_set_precise_purging(r, on_off);
+                        } else if (apply_mods) {
+                            replica_set_precise_purging(r, 0);
+                        }
+                    }
+                }
                 else
                 {
                     *returncode = LDAP_UNWILLING_TO_PERFORM;

ldap/servers/plugins/replication/repl_globals.c

@@ -118,6 +118,7 @@ const char *type_replicaAbortCleanRUV = "nsds5ReplicaAbortCleanRUV";
 const char *type_replicaProtocolTimeout = "nsds5ReplicaProtocolTimeout";
 const char *type_replicaBackoffMin = "nsds5ReplicaBackoffMin";
 const char *type_replicaBackoffMax = "nsds5ReplicaBackoffMax";
+const char *type_replicaPrecisePurge = "nsds5ReplicaPreciseTombstonePurging";
 
 /* Attribute names for replication agreement attributes */
 const char *type_nsds5ReplicaHost = "nsds5ReplicaHost";

ldap/servers/plugins/replication/urp_tombstone.c

@@ -75,7 +75,7 @@ get_tombstone_csn(const Slapi_Entry *entry, const CSN **delcsn)
 	PRBool ists = PR_FALSE;
 	if (is_tombstone_entry(entry)) {
 		ists = PR_TRUE;
-		*delcsn = _get_deletion_csn((Slapi_Entry *)entry); /* cast away const */
+		*delcsn = entry_get_deletion_csn((Slapi_Entry *)entry); /* cast away const */
 	}
 
 	return ists;

ldap/servers/slapd/back-ldbm/import-threads.c

@@ -110,6 +110,26 @@ static int import_generate_uniqueid(ImportJob *job, Slapi_Entry *e)
     return( rc );
 }
 
+/*
+ * Check if the tombstonecsn is missing, if so add it.
+ */
+static void
+import_generate_tombstone_csn(Slapi_Entry *e)
+{
+    if(e->e_flags & SLAPI_ENTRY_FLAG_TOMBSTONE) {
+        if (attrlist_find( e->e_attrs, SLAPI_ATTR_TOMBSTONE_CSN ) == NULL){
+            const CSN *tombstone_csn = NULL;
+            char tombstone_csnstr[CSN_STRSIZE];
+
+            /* Add the tombstone csn str */
+            if((tombstone_csn = entry_get_deletion_csn(e))){
+                csn_as_string(tombstone_csn, PR_FALSE, tombstone_csnstr);
+                slapi_entry_add_string(e, SLAPI_ATTR_TOMBSTONE_CSN, tombstone_csnstr);
+            }
+        }
+    }
+}
+
 
 /**********  BETTER LDIF PARSER  **********/
 
@@ -642,6 +662,8 @@ import_producer(void *param)
         if (g_get_global_lastmod()) {
             import_add_created_attrs(e);
         }
+        /* Add nsTombstoneCSN to tombstone entries unless it's already present */
+        import_generate_tombstone_csn(e);
 
         ep = import_make_backentry(e, id);
         if ((ep == NULL) || (ep->ep_entry == NULL)) {
@@ -2795,6 +2817,7 @@ import_worker(void *param)
     int is_objectclass_attribute;
     int is_nsuniqueid_attribute;
     int is_nscpentrydn_attribute;
+    int is_nstombstonecsn_attribute;
     void *attrlist_cursor;
     
     PR_ASSERT(NULL != info);
@@ -2822,6 +2845,8 @@ import_worker(void *param)
         (strcasecmp(info->index_info->name, SLAPI_ATTR_UNIQUEID) == 0);
     is_nscpentrydn_attribute =
         (strcasecmp(info->index_info->name, SLAPI_ATTR_NSCP_ENTRYDN) == 0);
+    is_nstombstonecsn_attribute =
+        (strcasecmp(info->index_info->name, SLAPI_ATTR_TOMBSTONE_CSN) == 0);
 
     if (1 != idl_get_idl_new()) {
         /* Is there substring indexing going on here ? */
@@ -3012,6 +3037,20 @@ import_worker(void *param)
                     }
                 }
             }
+            if(is_nstombstonecsn_attribute){
+                const CSN *tomb_csn = entry_get_deletion_csn(ep->ep_entry);
+                char tomb_csnstr[CSN_STRSIZE];
+
+                if(tomb_csn){
+                    csn_as_string(tomb_csn, PR_FALSE, tomb_csnstr);
+                    ret = index_addordel_string(be, SLAPI_ATTR_TOMBSTONE_CSN,
+                                  tomb_csnstr, ep->ep_id, BE_INDEX_ADD, NULL);
+                    if (0 != ret) {
+                        /* Something went wrong, eg disk filled up */
+                        goto error;
+                    }
+                }
+            }
         }
         import_decref_entry(ep);
         info->last_ID_processed = id;

ldap/servers/slapd/back-ldbm/index.c

@@ -393,8 +393,11 @@ index_addordel_entry(
     /* if we are adding a tombstone entry (see ldbm_add.c) */
     if ((flags & BE_INDEX_TOMBSTONE) && (flags & BE_INDEX_ADD))
     {
+        const CSN *tombstone_csn = NULL;
+        char deletion_csn_str[CSN_STRSIZE];
         Slapi_DN parent;
         Slapi_DN *sdn = slapi_entry_get_sdn(e->ep_entry);
+
         slapi_sdn_init(&parent);
         slapi_sdn_get_parent(sdn, &parent);
         /*
@@ -417,6 +420,16 @@ index_addordel_entry(
             ldbm_nasty(errmsg, 1021, result);
             return( result );
         }
+
+        if((tombstone_csn = entry_get_deletion_csn(e->ep_entry))){
+            csn_as_string(tombstone_csn, PR_FALSE, deletion_csn_str);
+            result = index_addordel_string(be, SLAPI_ATTR_TOMBSTONE_CSN, deletion_csn_str, e->ep_id, flags, txn);
+            if ( result != 0 ) {
+                ldbm_nasty(errmsg, 1021, result);
+                return( result );
+            }
+        }
+
         slapi_sdn_done(&parent);
         if (entryrdn_get_switch()) { /* subtree-rename: on */
             Slapi_Attr* attr;

ldap/servers/slapd/back-ldbm/ldbm_add.c

@@ -562,6 +562,7 @@ ldbm_back_add( Slapi_PBlock *pb )
 				addingentry->ep_id = slapi_entry_attr_get_ulong(addingentry->ep_entry,"entryid");
 				slapi_entry_attr_delete(addingentry->ep_entry, SLAPI_ATTR_VALUE_PARENT_UNIQUEID);
 				slapi_entry_delete_string(addingentry->ep_entry, SLAPI_ATTR_OBJECTCLASS, SLAPI_ATTR_VALUE_TOMBSTONE);
+				slapi_entry_attr_delete(addingentry->ep_entry, SLAPI_ATTR_TOMBSTONE_CSN);
 				/* Now also remove the nscpEntryDN */
 				if (slapi_entry_attr_delete(addingentry->ep_entry, SLAPI_ATTR_NSCP_ENTRYDN) != 0){
 					LDAPDebug(LDAP_DEBUG_REPL, "Resurrection of %s - Couldn't remove %s\n", dn, SLAPI_ATTR_NSCP_ENTRYDN, 0);
@@ -697,6 +698,26 @@ ldbm_back_add( Slapi_PBlock *pb )
 						slapi_entry_set_flag(addingentry->ep_entry,
 									SLAPI_ENTRY_FLAG_TOMBSTONE);
 					}
+					if (!slapi_entry_attr_hasvalue(addingentry->ep_entry,
+							SLAPI_ATTR_OBJECTCLASS, SLAPI_ATTR_VALUE_TOMBSTONE))
+					{
+						slapi_entry_add_string(addingentry->ep_entry,
+								SLAPI_ATTR_OBJECTCLASS, SLAPI_ATTR_VALUE_TOMBSTONE);
+						slapi_entry_set_flag(addingentry->ep_entry,
+								SLAPI_ENTRY_FLAG_TOMBSTONE);
+					}
+					if (attrlist_find( addingentry->ep_entry->e_attrs, SLAPI_ATTR_TOMBSTONE_CSN ) == NULL){
+						const CSN *tombstone_csn = NULL;
+						char tombstone_csnstr[CSN_STRSIZE];
+
+						/* Add the missing nsTombstoneCSN attribute to the tombstone */
+						if((tombstone_csn = entry_get_deletion_csn(addingentry->ep_entry))){
+							csn_as_string(tombstone_csn, PR_FALSE, tombstone_csnstr);
+							slapi_entry_add_string(addingentry->ep_entry, SLAPI_ATTR_TOMBSTONE_CSN,
+									tombstone_csnstr);
+						}
+					}
+
 					if (NULL != operation->o_params.p.p_add.parentuniqueid)
 					{
 						slapi_entry_add_string(addingentry->ep_entry,
@@ -744,7 +765,7 @@ ldbm_back_add( Slapi_PBlock *pb )
 				}
 				pid = parententry->ep_id;
 
-				/* We may need to adjust the DN since parent could be a resrected conflict entry... */
+				/* We may need to adjust the DN since parent could be a resurrected conflict entry... */
 				if (!slapi_sdn_isparent(slapi_entry_get_sdn_const(parententry->ep_entry),
 				                        slapi_entry_get_sdn_const(addingentry->ep_entry))) {
 					Slapi_DN adjustedsdn = {0};
@@ -790,7 +811,7 @@ ldbm_back_add( Slapi_PBlock *pb )
 
 			/* Tentatively add the entry to the cache.  We do this after adding any
 			 * operational attributes to ensure that the cache is sized correctly. */
-			if ( cache_add_tentative( &inst->inst_cache, addingentry, NULL )!= 0 )
+			if ( cache_add_tentative( &inst->inst_cache, addingentry, NULL ) != 0 )
 			{
 				LDAPDebug1Arg(LDAP_DEBUG_CACHE, "cache_add_tentative concurrency detected: %s\n",
 				              slapi_entry_get_dn_const(addingentry->ep_entry));
@@ -886,7 +907,11 @@ ldbm_back_add( Slapi_PBlock *pb )
 			goto error_return; 
 		}
 		if (is_resurect_operation) {
-			retval = index_addordel_string(be,SLAPI_ATTR_OBJECTCLASS,SLAPI_ATTR_VALUE_TOMBSTONE,addingentry->ep_id,BE_INDEX_DEL|BE_INDEX_EQUALITY,&txn);
+			const CSN *tombstone_csn = NULL;
+			char deletion_csn_str[CSN_STRSIZE];
+
+			retval = index_addordel_string(be,SLAPI_ATTR_OBJECTCLASS, SLAPI_ATTR_VALUE_TOMBSTONE,
+			                 addingentry->ep_id, BE_INDEX_DEL|BE_INDEX_EQUALITY, &txn);
 			if (DB_LOCK_DEADLOCK == retval) {
 				LDAPDebug( LDAP_DEBUG_ARGS, "add 2 DB_LOCK_DEADLOCK\n", 0, 0, 0 );
 				/* Retry txn */
@@ -903,9 +928,33 @@ ldbm_back_add( Slapi_PBlock *pb )
 				}
 				goto error_return; 
 			}
+
+			/* Need to update the nsTombstoneCSN index */
+			if((tombstone_csn = entry_get_deletion_csn(tombstoneentry->ep_entry))){
+				csn_as_string(tombstone_csn, PR_FALSE, deletion_csn_str);
+				retval = index_addordel_string(be, SLAPI_ATTR_TOMBSTONE_CSN, deletion_csn_str,
+								 tombstoneentry->ep_id, BE_INDEX_DEL|BE_INDEX_EQUALITY, &txn);
+				if (DB_LOCK_DEADLOCK == retval) {
+					LDAPDebug( LDAP_DEBUG_ARGS, "add 3 DB_LOCK_DEADLOCK\n", 0, 0, 0 );
+					/* Retry txn */
+					continue;
+				}
+				if (0 != retval) {
+					LDAPDebug(LDAP_DEBUG_TRACE, "index_addordel_string TOMBSTONE csn(%s), err=%d %s\n",
+							  slapi_entry_get_dn_const(tombstoneentry->ep_entry),
+							  retval, (msg = dblayer_strerror( retval )) ? msg : "");
+					ADD_SET_ERROR(ldap_result_code, LDAP_OPERATIONS_ERROR, retry_count);
+					if (LDBM_OS_ERR_IS_DISKFULL(retval)) {
+						disk_full = 1;
+						goto diskfull_return;
+					}
+					goto error_return;
+				}
+			}
+
 			retval = index_addordel_string(be,SLAPI_ATTR_UNIQUEID,slapi_entry_get_uniqueid(addingentry->ep_entry),addingentry->ep_id,BE_INDEX_DEL|BE_INDEX_EQUALITY,&txn);
 			if (DB_LOCK_DEADLOCK == retval) {
-				LDAPDebug( LDAP_DEBUG_ARGS, "add 3 DB_LOCK_DEADLOCK\n", 0, 0, 0 );
+				LDAPDebug( LDAP_DEBUG_ARGS, "add 4 DB_LOCK_DEADLOCK\n", 0, 0, 0 );
 				/* Retry txn */
 				continue;
 			}
@@ -926,7 +975,7 @@ ldbm_back_add( Slapi_PBlock *pb )
 			                               addingentry->ep_id,
 			                               BE_INDEX_DEL|BE_INDEX_EQUALITY, &txn);
 			if (DB_LOCK_DEADLOCK == retval) {
-				LDAPDebug( LDAP_DEBUG_ARGS, "add 4 DB_LOCK_DEADLOCK\n", 0, 0, 0 );
+				LDAPDebug( LDAP_DEBUG_ARGS, "add 5 DB_LOCK_DEADLOCK\n", 0, 0, 0 );
 				/* Retry txn */
 				continue;
 			}

ldap/servers/slapd/back-ldbm/ldbm_delete.c

@@ -102,12 +102,14 @@ ldbm_back_delete( Slapi_PBlock *pb )
 	int parent_switched = 0;
 	int myrc = 0;
 	PRUint64 conn_id;
+	const CSN *tombstone_csn = NULL;
+	char deletion_csn_str[CSN_STRSIZE];
 	int op_id;
+
 	if (slapi_pblock_get(pb, SLAPI_CONN_ID, &conn_id) < 0) {
 		conn_id = 0; /* connection is NULL */
 	}
-	slapi_pblock_get(pb, SLAPI_OPERATION_ID, &op_id);
-
+	slapi_pblock_get( pb, SLAPI_OPERATION_ID, &op_id);
 	slapi_pblock_get( pb, SLAPI_BACKEND, &be);
 	slapi_pblock_get( pb, SLAPI_PLUGIN_PRIVATE, &li );
 	slapi_pblock_get( pb, SLAPI_DELETE_TARGET_SDN, &sdnp );
@@ -629,10 +631,13 @@ ldbm_back_delete( Slapi_PBlock *pb )
 					/* The suffix entry has no parent */
 					slapi_entry_add_string(tombstone->ep_entry, SLAPI_ATTR_VALUE_PARENT_UNIQUEID, parentuniqueid);
 				}
+				if(opcsn){
+					csn_as_string(opcsn, PR_FALSE, deletion_csn_str);
+					slapi_entry_add_string(tombstone->ep_entry, SLAPI_ATTR_TOMBSTONE_CSN, deletion_csn_str);
+				}
 				slapi_entry_add_string(tombstone->ep_entry, SLAPI_ATTR_NSCP_ENTRYDN, slapi_sdn_get_ndn(&nscpEntrySDN));
 				tomb_value = slapi_value_new_string(SLAPI_ATTR_VALUE_TOMBSTONE);
-				value_update_csn(tomb_value, CSN_TYPE_VALUE_UPDATED,
-					operation_get_csn(operation));
+				value_update_csn(tomb_value, CSN_TYPE_VALUE_UPDATED, operation_get_csn(operation));
 				slapi_entry_add_value(tombstone->ep_entry, SLAPI_ATTR_OBJECTCLASS, tomb_value);
 				slapi_value_free(&tomb_value);
 				/* XXXggood above used to be: slapi_entry_add_string(tombstone->ep_entry, SLAPI_ATTR_OBJECTCLASS, SLAPI_ATTR_VALUE_TOMBSTONE); */
@@ -846,6 +851,8 @@ ldbm_back_delete( Slapi_PBlock *pb )
 			 * above, but we want it to remain in the nsUniqueID and nscpEntryDN indexes
 			 * and for objectclass=tombstone.
 			 */
+
+
 			retval = index_addordel_string(be, SLAPI_ATTR_OBJECTCLASS, 
 							SLAPI_ATTR_VALUE_TOMBSTONE,
 							tombstone->ep_id,BE_INDEX_ADD, &txn);
@@ -858,14 +865,42 @@ ldbm_back_delete( Slapi_PBlock *pb )
 			}
 			if (retval) {
 				LDAPDebug( LDAP_DEBUG_ANY,
-							"delete (adding %s) failed, err=%d %s\n",
-							SLAPI_ATTR_VALUE_TOMBSTONE, retval,
-							(msg = dblayer_strerror( retval )) ? msg : "" );
+				           "delete (adding %s) failed, err=%d %s\n",
+				           SLAPI_ATTR_VALUE_TOMBSTONE, retval,
+				           (msg = dblayer_strerror( retval )) ? msg : "" );
 				if (LDBM_OS_ERR_IS_DISKFULL(retval)) disk_full = 1;
 				DEL_SET_ERROR(ldap_result_code, 
 							  LDAP_OPERATIONS_ERROR, retry_count);
 				goto error_return;
 			}
+
+			/* Need to update the nsTombstoneCSN index */
+			if((tombstone_csn = entry_get_deletion_csn(tombstone->ep_entry))){
+				csn_as_string(tombstone_csn, PR_FALSE, deletion_csn_str);
+				retval = index_addordel_string(be, SLAPI_ATTR_TOMBSTONE_CSN,
+				                               deletion_csn_str, tombstone->ep_id,
+				                               BE_INDEX_ADD, &txn);
+				if (DB_LOCK_DEADLOCK == retval) {
+					LDAPDebug( LDAP_DEBUG_BACKLDBM,
+					           "delete tombstone csn(adding %s) DB_LOCK_DEADLOCK\n",
+					           deletion_csn_str, 0, 0 );
+					/* Retry txn */
+					continue;
+				}
+				if (0 != retval) {
+					LDAPDebug( LDAP_DEBUG_ANY,
+							   "delete tombsone csn(adding %s) failed, err=%d %s\n",
+							   deletion_csn_str,
+							   retval,
+							   (msg = dblayer_strerror( retval )) ? msg : "" );
+					if (LDBM_OS_ERR_IS_DISKFULL(retval)){
+						disk_full = 1;
+					}
+					DEL_SET_ERROR(ldap_result_code, LDAP_OPERATIONS_ERROR, retry_count);
+					goto error_return;
+				}
+			}
+
 			retval = index_addordel_string(be, SLAPI_ATTR_UNIQUEID,
 							slapi_entry_get_uniqueid(tombstone->ep_entry),
 							tombstone->ep_id,BE_INDEX_ADD,&txn);
@@ -1002,7 +1037,7 @@ ldbm_back_delete( Slapi_PBlock *pb )
 		{
 			/* 
 			 * We need to remove the Tombstone entry from the remaining indexes:
-			 * objectclass=nsTombstone, nsUniqueID, nscpEntryDN
+			 * objectclass=nsTombstone, nsUniqueID, nscpEntryDN, nsTombstoneCSN
 			 */
 			char *nscpedn = NULL;
 
@@ -1026,6 +1061,34 @@ ldbm_back_delete( Slapi_PBlock *pb )
 							  LDAP_OPERATIONS_ERROR, retry_count);
 				goto error_return;
 			}
+
+			/* Need to update the nsTombstoneCSN index */
+			if((tombstone_csn = entry_get_deletion_csn(e->ep_entry))){
+				csn_as_string(tombstone_csn, PR_FALSE, deletion_csn_str);
+				retval = index_addordel_string(be, SLAPI_ATTR_TOMBSTONE_CSN,
+				                               deletion_csn_str, e->ep_id,
+				                               BE_INDEX_DEL|BE_INDEX_EQUALITY, &txn);
+				if (DB_LOCK_DEADLOCK == retval) {
+					LDAPDebug( LDAP_DEBUG_BACKLDBM,
+					           "delete tombstone csn(deleting %s) DB_LOCK_DEADLOCK\n",
+					           deletion_csn_str, 0, 0 );
+					/* Retry txn */
+					continue;
+				}
+				if (0 != retval) {
+					LDAPDebug( LDAP_DEBUG_ANY,
+					           "delete tombsone csn(deleting %s) failed, err=%d %s\n",
+					           deletion_csn_str,
+					           retval,
+					           (msg = dblayer_strerror( retval )) ? msg : "" );
+					if (LDBM_OS_ERR_IS_DISKFULL(retval)){
+						disk_full = 1;
+					}
+					DEL_SET_ERROR(ldap_result_code, LDAP_OPERATIONS_ERROR, retry_count);
+					goto error_return;
+				}
+			}
+
 			retval = index_addordel_string(be, SLAPI_ATTR_UNIQUEID,
 							slapi_entry_get_uniqueid(e->ep_entry),
 							e->ep_id, BE_INDEX_DEL|BE_INDEX_EQUALITY, &txn);

ldap/servers/slapd/back-ldbm/ldbm_modify.c

@@ -418,15 +418,18 @@ ldbm_back_modify( Slapi_PBlock *pb )
 	int opreturn = 0;
 	int mod_count = 0;
 	int not_an_error = 0;
+	int fixup_tombstone = 0;
 
 	slapi_pblock_get( pb, SLAPI_BACKEND, &be);
 	slapi_pblock_get( pb, SLAPI_PLUGIN_PRIVATE, &li );
 	slapi_pblock_get( pb, SLAPI_TARGET_ADDRESS, &addr );
 	slapi_pblock_get( pb, SLAPI_MODIFY_MODS, &mods );
 	slapi_pblock_get( pb, SLAPI_TXN, (void**)&parent_txn );
-	slapi_pblock_get (pb, SLAPI_IS_REPLICATED_OPERATION, &repl_op);
-
+	slapi_pblock_get( pb, SLAPI_IS_REPLICATED_OPERATION, &repl_op);
 	slapi_pblock_get( pb, SLAPI_OPERATION, &operation );
+
+	fixup_tombstone = operation_is_flag_set(operation, OP_FLAG_TOMBSTONE_FIXUP);
+
 	dblayer_txn_init(li,&txn); /* must do this before first goto error_return */
 	/* the calls to perform searches require the parent txn if any
 	   so set txn to the parent_txn until we begin the child transaction */
@@ -566,13 +569,15 @@ ldbm_back_modify( Slapi_PBlock *pb )
 				}
 			}
 		
-			if ( !is_fixup_operation )
+			if ( !is_fixup_operation && !fixup_tombstone)
 			{
-				if (!repl_op && slapi_entry_flag_is_set(e->ep_entry, SLAPI_ENTRY_FLAG_TOMBSTONE) ) {
+				if (!repl_op && slapi_entry_flag_is_set(e->ep_entry, SLAPI_ENTRY_FLAG_TOMBSTONE))
+				{
 					ldap_result_code = LDAP_UNWILLING_TO_PERFORM;
                 			ldap_result_message = "Operation not allowed on tombstone entry.";
 					slapi_log_error(SLAPI_LOG_FATAL, "ldbm_back_modify",
-						"Attempt to modify a tombstone entry %s\n", slapi_sdn_get_dn(slapi_entry_get_sdn_const( e->ep_entry )));
+						"Attempt to modify a tombstone entry %s\n",
+						slapi_sdn_get_dn(slapi_entry_get_sdn_const( e->ep_entry )));
 					goto error_return;
 				}
 				opcsn = operation_get_csn (operation);

ldap/servers/slapd/back-ldbm/ldbm_modrdn.c

@@ -732,6 +732,7 @@ ldbm_back_modrdn( Slapi_PBlock *pb )
                 slapi_log_error(SLAPI_LOG_REPL, "ldbm_back_modrdn",
                                 "Resurrecting an entry %s\n", slapi_entry_get_dn(ec->ep_entry));
                 slapi_entry_attr_delete(ec->ep_entry, SLAPI_ATTR_VALUE_PARENT_UNIQUEID);
+                slapi_entry_attr_delete(ec->ep_entry, SLAPI_ATTR_TOMBSTONE_CSN);
                 slapi_entry_delete_string(ec->ep_entry, SLAPI_ATTR_OBJECTCLASS, SLAPI_ATTR_VALUE_TOMBSTONE);
                 /* Now also remove the nscpEntryDN */
                 if (slapi_entry_attr_delete(ec->ep_entry, SLAPI_ATTR_NSCP_ENTRYDN) != 0){

ldap/servers/slapd/back-ldbm/ldif2ldbm.c

@@ -2178,6 +2178,69 @@ ldbm_back_ldbm2index(Slapi_PBlock *pb)
             continue;
         }
 
+        /*
+         * If this entry is a tombstone, update the 'nstombstonecsn' index
+         */
+        if(ep->ep_entry->e_flags & SLAPI_ENTRY_FLAG_TOMBSTONE){
+            const CSN *tombstone_csn = NULL;
+            char deletion_csn_str[CSN_STRSIZE];
+
+            if((tombstone_csn = entry_get_deletion_csn(ep->ep_entry))){
+                if (!run_from_cmdline) {
+                    rc = dblayer_txn_begin(be, NULL, &txn);
+                    if (0 != rc) {
+                        slapi_log_error(SLAPI_LOG_FATAL, "ldbm_back_ldbm2index",
+                            "%s: ERROR: failed to begin txn for update index '%s' (err %d: %s)\n",
+                            inst->inst_name, SLAPI_ATTR_TOMBSTONE_CSN, rc, dblayer_strerror(rc));
+                        if (task) {
+                            slapi_task_log_notice(task, "%s: ERROR: failed to begin txn for "
+                                                  "update index '%s' (err %d: %s)",
+                                                  inst->inst_name, indexAttrs[j], rc,
+                                                  dblayer_strerror(rc));
+                        }
+                        return_value = -2;
+                        goto err_out;
+                    }
+                }
+
+                csn_as_string(tombstone_csn, PR_FALSE, deletion_csn_str);
+                rc = index_addordel_string(be, SLAPI_ATTR_TOMBSTONE_CSN, deletion_csn_str,
+                                           ep->ep_id, BE_INDEX_ADD, &txn);
+                if (rc != 0) {
+                    slapi_log_error(SLAPI_LOG_FATAL, "ldbm_back_ldbm2index",
+                        "%s: ERROR: failed to update index '%s' (err %d: %s)\n",
+                        inst->inst_name, SLAPI_ATTR_TOMBSTONE_CSN, rc, dblayer_strerror(rc));
+                    if (task) {
+                        slapi_task_log_notice(task, "%s: ERROR: failed to update index '%s' "
+                                                    "(err %d: %s)", inst->inst_name,
+                                                    SLAPI_ATTR_TOMBSTONE_CSN, rc,
+                                                    dblayer_strerror(rc));
+                    }
+                    if (!run_from_cmdline) {
+                        dblayer_txn_abort(be, &txn);
+                    }
+                    return_value = -2;
+                    goto err_out;
+                }
+                if (!run_from_cmdline) {
+                    rc = dblayer_txn_commit(be, &txn);
+                    if (0 != rc) {
+                        slapi_log_error(SLAPI_LOG_FATAL, "ldbm_back_ldbm2index",
+                            "%s: ERROR: failed to commit txn for update index '%s' (err %d: %s)\n",
+                            inst->inst_name, SLAPI_ATTR_TOMBSTONE_CSN, rc, dblayer_strerror(rc));
+                        if (task) {
+                            slapi_task_log_notice(task,"%s: ERROR: failed to commit txn for "
+                                                  "update index '%s' (err %d: %s)",
+                                                  inst->inst_name, SLAPI_ATTR_TOMBSTONE_CSN,
+                                                  rc, dblayer_strerror(rc));
+                        }
+                        return_value = -2;
+                        goto err_out;
+                    }
+                }
+            }
+        }
+
         /*
          * Update the attribute indexes
          */

ldap/servers/slapd/entrywsi.c

@@ -1527,3 +1527,26 @@ resolve_attribute_state_single_valued(Slapi_Entry *e, Slapi_Attr *a, int attribu
 	}
 }
 
+const CSN *
+entry_get_deletion_csn(Slapi_Entry *e)
+{
+	const CSN *deletion_csn = NULL;
+
+	PR_ASSERT(NULL != e);
+	if (NULL != e)
+	{
+		Slapi_Attr *oc_attr = NULL;
+		if (entry_attr_find_wsi(e, SLAPI_ATTR_OBJECTCLASS, &oc_attr) == ATTRIBUTE_PRESENT)
+		{
+			Slapi_Value *tombstone_value = NULL;
+			struct berval v;
+			v.bv_val = SLAPI_ATTR_VALUE_TOMBSTONE;
+			v.bv_len = strlen(SLAPI_ATTR_VALUE_TOMBSTONE);
+			if (attr_value_find_wsi(oc_attr, &v, &tombstone_value) == VALUE_PRESENT)
+			{
+				deletion_csn = value_get_csn(tombstone_value, CSN_TYPE_VALUE_UPDATED);
+			}
+		}
+	}
+	return deletion_csn;
+}

ldap/servers/slapd/mapping_tree.c

@@ -2174,6 +2174,7 @@ int slapi_mapping_tree_select(Slapi_PBlock *pb, Slapi_Backend **be, Slapi_Entry
     int ret;
     int scope=LDAP_SCOPE_BASE;
     int op_type;
+    int fixup = 0;
     
 
     if(mapping_tree_freed){
@@ -2192,6 +2193,7 @@ int slapi_mapping_tree_select(Slapi_PBlock *pb, Slapi_Backend **be, Slapi_Entry
 
     /* Get the target for this op */
     target_sdn = operation_get_target_spec (op);
+    fixup = operation_is_flag_set(op, OP_FLAG_TOMBSTONE_FIXUP);
 
     if(!mapping_tree_inited) {
         mapping_tree_init();
@@ -2240,14 +2242,17 @@ int slapi_mapping_tree_select(Slapi_PBlock *pb, Slapi_Backend **be, Slapi_Entry
      * or if the whole server is readonly AND backend is public (!private)
      */
     if ((ret == LDAP_SUCCESS) && *be && !be_isdeleted(*be) &&
-        ((*be)->be_readonly ||
-        (slapi_config_get_readonly() && !slapi_be_private(*be)))) {
+        (((*be)->be_readonly && !fixup) ||
+         ((slapi_config_get_readonly() && !fixup) &&
+         !slapi_be_private(*be))) )
+    {
         unsigned long op_type = operation_get_type(op);
 
         if ((op_type != SLAPI_OPERATION_SEARCH) &&
             (op_type != SLAPI_OPERATION_COMPARE) &&
             (op_type != SLAPI_OPERATION_BIND) && 
-            (op_type != SLAPI_OPERATION_UNBIND)) {
+            (op_type != SLAPI_OPERATION_UNBIND))
+        {
             ret = LDAP_UNWILLING_TO_PERFORM;
             PL_strncpyz(errorbuf, slapi_config_get_readonly() ? 
                     "Server is read-only" :

ldap/servers/slapd/slapi-plugin.h

@@ -498,6 +498,7 @@ NSPR_API(PRUint32) PR_fprintf(struct PRFileDesc* fd, const char *fmt, ...)
 #define SLAPI_ATTR_UNIQUEID			"nsuniqueid"
 #define SLAPI_ATTR_OBJECTCLASS			"objectclass"
 #define SLAPI_ATTR_VALUE_TOMBSTONE		"nsTombstone"
+#define SLAPI_ATTR_TOMBSTONE_CSN		"nsTombstoneCSN"
 #define SLAPI_ATTR_VALUE_PARENT_UNIQUEID	"nsParentUniqueID"
 #define SLAPI_ATTR_VALUE_SUBENTRY		"ldapsubentry"
 #define SLAPI_ATTR_NSCP_ENTRYDN 		"nscpEntryDN"
@@ -508,6 +509,7 @@ NSPR_API(PRUint32) PR_fprintf(struct PRFileDesc* fd, const char *fmt, ...)
 #define SLAPI_ATTR_UNIQUEID_LENGTH		10
 #define SLAPI_ATTR_OBJECTCLASS_LENGTH		11
 #define SLAPI_ATTR_VALUE_TOMBSTONE_LENGTH	11
+#define SLAPI_ATTR_TOMBSTONE_CSN_LENGTH		14
 #define SLAPI_ATTR_VALUE_PARENT_UNIQUEID_LENGTH	16
 #define SLAPI_ATTR_VALUE_SUBENTRY_LENGTH	12
 #define SLAPI_ATTR_NSCP_ENTRYDN_LENGTH 		11

ldap/servers/slapd/slapi-private.h

@@ -352,6 +352,7 @@ void entry_add_rdn_csn(Slapi_Entry *e, const CSN *csn);
 /* this adds a csn to the entry's e_dncsnset but makes sure the set is in increasing csn order */
 #define ENTRY_DNCSN_INCREASING 0x1 /* for flags below */
 int entry_add_dncsn_ext(Slapi_Entry *entry, const CSN *csn, PRUint32 flags);
+const CSN *entry_get_deletion_csn(Slapi_Entry *entry);
 
 /* attr.c */
 Slapi_Attr *slapi_attr_init_locking_optional(Slapi_Attr *a, const char *type, PRBool use_lock);
@@ -442,7 +443,8 @@ char *slapi_filter_to_string_internal( const struct slapi_filter *f, char *buf,
 #define OP_FLAG_PAGED_RESULTS            0x040000 /* simple paged results */
 #define OP_FLAG_SERVER_SIDE_SORTING      0x080000 /* server side sorting  */
 #define OP_FLAG_REVERSE_CANDIDATE_ORDER  0x100000 /* reverse the search candidate list */
-#define OP_FLAG_NEVER_CACHE		 0x200000 /* never keep the entry in cache */
+#define OP_FLAG_NEVER_CACHE              0x200000 /* never keep the entry in cache */
+#define OP_FLAG_TOMBSTONE_FIXUP          0x400000 /* operation is tombstone fixup op */
 
 /* reverse search states */
 #define REV_STARTED 1

ldap/servers/slapd/task.c

@@ -67,6 +67,7 @@ static int shutting_down = 0;
 #define TASK_RESTORE_DN   "cn=restore,cn=tasks,cn=config"
 #define TASK_INDEX_DN     "cn=index,cn=tasks,cn=config"
 #define TASK_UPGRADEDB_DN "cn=upgradedb,cn=tasks,cn=config"
+#define TASK_TOMBSTONE_FIXUP_DN  "cn=fixup tombstones,cn=tasks,cn=config"
 
 #define TASK_LOG_NAME           "nsTaskLog"
 #define TASK_STATUS_NAME        "nsTaskStatus"
@@ -77,6 +78,10 @@ static int shutting_down = 0;
 #define DEFAULT_TTL     "120"   /* seconds */
 #define TASK_SYSCONFIG_FILE_ATTR "sysconfigfile" /* sysconfig reload task file attr */
 #define TASK_SYSCONFIG_LOGCHANGES_ATTR "logchanges"
+#define TASK_TOMBSTONE_FIXUP "fixup tombstones task"
+#define TASK_TOMBSTONE_FIXUP_BACKEND "backend"
+#define TASK_TOMBSTONE_FIXUP_SUFFIX "suffix"
+#define TASK_TOMBSTONE_FIXUP_STRIPCSN "stripcsn"
 
 #define LOG_BUFFER              256
 /* if the cumul. log gets larger than this, it's truncated: */
@@ -224,29 +229,12 @@ void slapi_task_log_status(Slapi_Task *task, char *format, ...)
     slapi_task_status_changed(task);
 }
 
-void slapi_task_log_status_ext(Slapi_Task *task, char *format, va_list ap)
-{
-    if (! task->task_status)
-        task->task_status = (char *)slapi_ch_malloc(10 * LOG_BUFFER);
-    if (! task->task_status)
-        return;        /* out of memory? */
-
-    PR_vsnprintf(task->task_status, (10 * LOG_BUFFER), format, ap);
-    slapi_task_status_changed(task);
-}
-
-/* this adds a line to the 'nsTaskLog' value, which is cumulative (anything
- * logged here is added to the end)
- */
-void slapi_task_log_notice(Slapi_Task *task, char *format, ...)
+void slapi_task_log_notice_ext(Slapi_Task *task, char *format, va_list ap)
 {
-    va_list ap;
     char buffer[LOG_BUFFER];
     size_t len;
 
-    va_start(ap, format);
     PR_vsnprintf(buffer, LOG_BUFFER, format, ap);
-    va_end(ap);
 
     if (task->task_log_lock) {
         PR_Lock(task->task_log_lock);
@@ -286,12 +274,29 @@ void slapi_task_log_notice(Slapi_Task *task, char *format, ...)
     slapi_task_status_changed(task);
 }
 
-void slapi_task_log_notice_ext(Slapi_Task *task, char *format, va_list ap)
+void slapi_task_log_status_ext(Slapi_Task *task, char *format, va_list ap)
+{
+    if (! task->task_status)
+        task->task_status = (char *)slapi_ch_malloc(10 * LOG_BUFFER);
+    if (! task->task_status)
+        return;        /* out of memory? */
+
+    PR_vsnprintf(task->task_status, (10 * LOG_BUFFER), format, ap);
+    slapi_task_status_changed(task);
+}
+
+/* this adds a line to the 'nsTaskLog' value, which is cumulative (anything
+ * logged here is added to the end)
+ */
+void slapi_task_log_notice(Slapi_Task *task, char *format, ...)
 {
+    va_list ap;
     char buffer[LOG_BUFFER];
     size_t len;
 
+    va_start(ap, format);
     PR_vsnprintf(buffer, LOG_BUFFER, format, ap);
+    va_end(ap);
 
     if (task->task_log_lock) {
         PR_Lock(task->task_log_lock);
@@ -2090,6 +2095,314 @@ done:
     return rc;
 }
 
+/*
+ * Add the nsTombstoneCSN attribute/value to the entry.
+ */
+static int
+fixup_tombstone(Slapi_PBlock *pb, char *suffix, Slapi_Entry *e, int *fixup_count)
+{
+    LDAPMod mod;
+    LDAPMod *mods[2];
+    const CSN *tombstone_csn = NULL;
+    char deletion_csn_str[CSN_STRSIZE];
+    char *val[2];
+    int rc = LDAP_SUCCESS;
+
+    if((tombstone_csn = entry_get_deletion_csn(e))){
+        slapi_log_error(SLAPI_LOG_REPL, TASK_TOMBSTONE_FIXUP,
+            "Fixing tombstone (%s)\n", slapi_entry_get_dn(e));
+
+        /* We have an entry tombstone that needs fixing */
+        slapi_pblock_init(pb);
+        csn_as_string(tombstone_csn, PR_FALSE, deletion_csn_str);
+        mods[0] = &mod;
+        mods[1] = 0;
+
+        val[0] = deletion_csn_str;
+        val[1] = 0;
+
+        mod.mod_op = LDAP_MOD_ADD;
+        mod.mod_type = SLAPI_ATTR_TOMBSTONE_CSN;
+        mod.mod_values = val;
+
+        slapi_modify_internal_set_pb_ext( pb, slapi_entry_get_sdn(e),
+            mods, 0, 0, (void *)plugin_get_default_component_id(),
+            OP_FLAG_TOMBSTONE_ENTRY | OP_FLAG_TOMBSTONE_FIXUP);
+        slapi_modify_internal_pb(pb);
+
+        slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc);
+        pblock_done(pb);
+        if(rc == LDAP_SUCCESS){
+            (*fixup_count)++;
+        }
+    }
+
+    return rc;
+}
+
+/*
+ * Strip out nsTombstoneCSN so the task can be run again to remove them.  Used
+ * solely for testing the fixup task.
+ */
+static void
+strip_tombstone(Slapi_PBlock *pb, char *suffix, Slapi_Entry *e, int *strip_count)
+{
+    LDAPMod mod;
+    LDAPMod *mods[2];
+    int rc = 0;
+
+    slapi_log_error(SLAPI_LOG_REPL, TASK_TOMBSTONE_FIXUP,
+            "Stripping tombstone (%s)\n", slapi_entry_get_dn(e));
+
+    /* We have an entry tombstone that needs stripping */
+    slapi_pblock_init(pb);
+    mods[0] = &mod;
+    mods[1] = 0;
+
+    mod.mod_op = LDAP_MOD_DELETE;
+    mod.mod_type = SLAPI_ATTR_TOMBSTONE_CSN;
+    mod.mod_values = NULL;
+
+    slapi_modify_internal_set_pb_ext( pb, slapi_entry_get_sdn(e),
+             mods, 0, 0, (void *)plugin_get_default_component_id(),
+             OP_FLAG_TOMBSTONE_ENTRY | OP_FLAG_TOMBSTONE_FIXUP);
+    slapi_modify_internal_pb(pb);
+
+    slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc);
+    pblock_done(pb);
+
+    if(rc == LDAP_SUCCESS){
+        (*strip_count)++;
+    } else {
+        slapi_log_error(SLAPI_LOG_REPL, TASK_TOMBSTONE_FIXUP,
+            "Stripping tombstone (%s) failed, error %d\n", slapi_entry_get_dn(e), rc);
+    }
+}
+
+struct task_tombstone_data
+{
+    char **base;
+    int stripcsn;
+    Slapi_Task *task;
+};
+
+/*
+ * Fix tombstone thread - add missing nsTombstoneCSN
+ */
+static void
+task_fixup_tombstone_thread(void *arg)
+{
+    struct task_tombstone_data *task_data = arg;
+    Slapi_Entry **entries = NULL;
+    Slapi_Task *task = task_data->task;
+    char **base = task_data->base;
+    char *filter = NULL;
+    int fixup_count = 0;
+    int rc, i, j;
+
+    slapi_task_begin(task, 1);
+    slapi_task_log_notice(task, "Beginning tombstone fixup task...\n");
+    slapi_log_error(SLAPI_LOG_REPL, TASK_TOMBSTONE_FIXUP,
+                    "Beginning tombstone fixup task...\n");
+
+    if(task_data->stripcsn){
+        /* find tombstones with nsTombstoneCSN */
+        filter = "(&(nstombstonecsn=*)(objectclass=nsTombstone))";
+    } else {
+        /* find tombstones missing nsTombstoneCSN */
+        filter = "(&(!(nstombstonecsn=*))(objectclass=nsTombstone))";
+    }
+
+    /* Okay check the specified backends only */
+    for(i = 0; base && base[i]; i++){
+        Slapi_PBlock *search_pb = slapi_pblock_new();
+
+        /* find entries that need fixing... */
+        slapi_search_internal_set_pb(search_pb, base[i], LDAP_SCOPE_SUBTREE,
+                filter, NULL, 0, NULL, NULL, plugin_get_default_component_id(), 0);
+        slapi_search_internal_pb(search_pb);
+
+        slapi_pblock_get(search_pb, SLAPI_PLUGIN_INTOP_RESULT, &rc);
+        if (rc != LDAP_SUCCESS) {
+            slapi_task_log_notice(task,
+                    "Failed to search backend for tombstones, error %d\n", rc);
+            slapi_log_error(SLAPI_LOG_REPL, TASK_TOMBSTONE_FIXUP,
+                    "Failed to search backend for tombstones, error %d\n", rc);
+            slapi_pblock_destroy(search_pb);
+            slapi_task_finish(task, rc);
+            return;
+        }
+
+        slapi_pblock_get(search_pb, SLAPI_PLUGIN_INTOP_SEARCH_ENTRIES, &entries);
+        if (entries) {
+            Slapi_PBlock *fixup_pb = slapi_pblock_new();
+
+            /* process all the tombstone entries */
+            for (j = 0; entries[j]; j++){
+                if(task_data->stripcsn){
+                    /* strip nsTombstoneCSN - used to testing */
+                    strip_tombstone(fixup_pb, base[i], entries[j], &fixup_count);
+                } else if((rc = fixup_tombstone(fixup_pb, base[i], entries[j], &fixup_count))){
+                    /* Failed to update tombstone, log it and move on... */
+                    slapi_task_log_notice(task,
+                           "Failed to update tombstone entry (%s) error %d\n",
+                            slapi_entry_get_dn(entries[j]), rc);
+                    slapi_log_error(SLAPI_LOG_FATAL, TASK_TOMBSTONE_FIXUP,
+                            "Failed to update tombstone entry (%s) error %d\n",
+                            slapi_entry_get_dn(entries[j]), rc);
+                }
+            }
+            slapi_free_search_results_internal(search_pb);
+            slapi_pblock_destroy(fixup_pb);
+        }
+        slapi_pblock_destroy(search_pb);
+        slapi_task_inc_progress(task);
+    }
+    slapi_task_log_notice(task, "%s %d tombstones.\n",
+                          task_data->stripcsn ? "Stripped" : "Fixed", fixup_count);
+    slapi_log_error(SLAPI_LOG_REPL, TASK_TOMBSTONE_FIXUP, "%s %d tombstones.\n",
+                    task_data->stripcsn ? "Stripped" : "Fixed", fixup_count);
+    slapi_task_inc_progress(task);
+    slapi_task_finish(task, rc);
+    slapi_ch_array_free(base);
+    slapi_ch_free((void **)&task_data);
+}
+
+
+/*
+ *  task_fixup_tombstones_add
+ *
+ *  Check all the existing tombstones and add nsTombstoneCSN if missing.
+ *
+ *  dn: cn=fixem,cn=fixup tombstones,cn=tasks,cn=config
+ *  objectclass: top
+ *  objectclass: extensibleObject
+ *  cn: fixem
+ *  backend: userRoot
+ *  suffix: dc=example,dc=com
+ *  stripcsn: yes
+ *
+ *  backend & suffix are optional.  If skipped, all backends/suffixes are
+ *  checked.  Multiple suffixes can also be specified.
+ *
+ *  Hidden option: "stripcsn" is strictly used to verify the fixup task: run
+ *  the task using the strip option to strip tombstones of "nsTombstoneCSN",
+ *  then run task, without the strip option, to add "nsTombstoneCSN" back.
+ */
+static int
+task_fixup_tombstones_add(Slapi_PBlock *pb, Slapi_Entry *e, Slapi_Entry *eAfter,
+                          int *returncode, char *returntext, void *arg)
+{
+    Slapi_Backend *be = NULL;
+    Slapi_Task *task = NULL;
+    struct task_tombstone_data *task_data = NULL;
+    const Slapi_DN *base_sdn = NULL;
+    PRThread *thread = NULL;
+    char **backend = NULL;
+    char **suffix = NULL;
+    char **base = NULL;
+    char *stripcsn = NULL;
+    int i;
+
+    /*
+     * Get the task options.  We will store all the "backends" in the suffix array.
+     */
+    if((suffix = slapi_entry_attr_get_charray(e, TASK_TOMBSTONE_FIXUP_SUFFIX))){
+        for (i = 0; suffix && suffix[i]; i++){
+            char *dn = slapi_create_dn_string("%s", suffix[i]);
+
+            if(dn){
+                if(slapi_dn_syntax_check(pb, dn, 1)){
+                    /* invalid suffix name */
+                    PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE,
+                        "Invalid DN (%s) used for \"suffix\"\n", suffix[i]);
+                    *returncode = LDAP_INVALID_DN_SYNTAX;
+                    goto done;
+                } else {
+                    slapi_ch_array_add(&base, dn);
+                }
+            }
+        }
+    }
+    if((backend = slapi_entry_attr_get_charray(e, TASK_TOMBSTONE_FIXUP_BACKEND))){
+        for (i = 0; backend && backend[i]; i++){
+            if((be = slapi_be_select_by_instance_name(backend[i]))){
+                if((base_sdn = slapi_be_getsuffix(be, 0))){
+                    slapi_ch_array_add(&base, slapi_ch_strdup(slapi_sdn_get_ndn(base_sdn)));
+                } else {
+                    /* failed to get a suffix */
+                    PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE,
+                            "Failed to find a suffix for the backend(%s)\n", backend[i]);
+                    *returncode = LDAP_UNWILLING_TO_PERFORM;
+                    goto done;
+                }
+            } else {
+                /* Failed to find a backend */
+                PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE,
+                        "Failed to find a backend using (%s)\n", backend[i]);
+                *returncode = LDAP_UNWILLING_TO_PERFORM;
+                goto done;
+            }
+        }
+    }
+
+    /*
+     * If suffix is NULL, we check all the backends
+     */
+    if(base == NULL){
+        char *cookie = NULL;
+
+        /* Gather all the backends */
+        be = slapi_get_first_backend(&cookie);
+        while(be){
+            if((base_sdn = slapi_be_getsuffix(be, 0)) && !be->be_private){
+                const char *suf = slapi_sdn_get_ndn(base_sdn);
+                /* Need to skip the retro changelog */
+                if(strcmp(suf, "cn=changelog"))
+                {
+                    slapi_ch_array_add(&base, slapi_ch_strdup(suf));
+                }
+            }
+            be = slapi_get_next_backend(cookie);
+        }
+    }
+
+    task = slapi_new_task(slapi_entry_get_ndn(e));
+    task_data = (struct task_tombstone_data *)slapi_ch_calloc(1, sizeof(struct task_tombstone_data));
+    task_data->base = base;
+    task_data->task = task;
+    if((stripcsn = slapi_entry_attr_get_charptr(e, TASK_TOMBSTONE_FIXUP_STRIPCSN))){
+        if(strcasecmp(stripcsn, "yes") == 0 || strcasecmp(stripcsn, "on") == 0){
+            task_data->stripcsn = 1;
+        }
+        slapi_ch_free_string(&stripcsn);
+    }
+
+    /* start the db2index as a separate thread */
+    thread = PR_CreateThread(PR_USER_THREAD, task_fixup_tombstone_thread,
+                             (void *)task_data, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
+                             PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);
+    if (thread == NULL) {
+        LDAPDebug(LDAP_DEBUG_ANY,
+                  "task_fixup_tombstones_add: unable to create index thread!\n", 0, 0, 0);
+        *returncode = LDAP_OPERATIONS_ERROR;
+        slapi_task_finish(task, *returncode);
+        slapi_ch_array_free(base);
+        slapi_ch_free((void **)&task_data);
+        return SLAPI_DSE_CALLBACK_ERROR;
+    }
+
+done:
+    slapi_ch_array_free(suffix);
+    slapi_ch_array_free(backend);
+
+    if (*returncode != LDAP_SUCCESS){
+        return SLAPI_DSE_CALLBACK_ERROR;
+    }
+
+    return SLAPI_DSE_CALLBACK_OK;
+}
+
 /* cleanup old tasks that may still be in the DSE from a previous session
  * (this can happen if the server crashes [no matter how unlikely we like
  * to think that is].)
@@ -2170,6 +2483,7 @@ void task_init(void)
     slapi_task_register_handler("index", task_index_add);
     slapi_task_register_handler("upgradedb", task_upgradedb_add);
     slapi_task_register_handler("sysconfig reload", task_sysconfig_reload_add);
+    slapi_task_register_handler("fixup tombstones", task_fixup_tombstones_add);
 }
 
 /* called when the server is shutting down -- abort all existing tasks */