2 Unix SMB/CIFS mplementation.
4 DSDB replication service periodic notification handling
6 Copyright (C) Andrew Tridgell 2009
7 based on drepl_periodic
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation; either version 3 of the License, or
12 (at your option) any later version.
14 This program is distributed in the hope that it will be useful,
15 but WITHOUT ANY WARRANTY; without even the implied warranty of
16 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 GNU General Public License for more details.
19 You should have received a copy of the GNU General Public License
20 along with this program. If not, see <http://www.gnu.org/licenses/>.
25 #include "lib/events/events.h"
26 #include "dsdb/samdb/samdb.h"
27 #include "auth/auth.h"
28 #include "smbd/service.h"
29 #include "lib/messaging/irpc.h"
30 #include "dsdb/repl/drepl_service.h"
31 #include "lib/ldb/include/ldb_errors.h"
32 #include "../lib/util/dlinklist.h"
33 #include "librpc/gen_ndr/ndr_misc.h"
34 #include "librpc/gen_ndr/ndr_drsuapi.h"
35 #include "librpc/gen_ndr/ndr_drsblobs.h"
36 #include "libcli/composite/composite.h"
37 #include "../lib/util/tevent_ntstatus.h"
40 struct dreplsrv_op_notify_state
{
41 struct dreplsrv_notify_operation
*op
;
45 static void dreplsrv_op_notify_connect_done(struct tevent_req
*subreq
);
48 start the ReplicaSync async call
50 static struct tevent_req
*dreplsrv_op_notify_send(TALLOC_CTX
*mem_ctx
,
51 struct tevent_context
*ev
,
52 struct dreplsrv_notify_operation
*op
)
54 struct tevent_req
*req
;
55 struct dreplsrv_op_notify_state
*state
;
56 struct tevent_req
*subreq
;
58 req
= tevent_req_create(mem_ctx
, &state
,
59 struct dreplsrv_op_notify_state
);
65 subreq
= dreplsrv_out_drsuapi_send(state
,
67 op
->source_dsa
->conn
);
68 if (tevent_req_nomem(subreq
, req
)) {
69 return tevent_req_post(req
, ev
);
71 tevent_req_set_callback(subreq
, dreplsrv_op_notify_connect_done
, req
);
76 static void dreplsrv_op_notify_replica_sync_trigger(struct tevent_req
*req
);
78 static void dreplsrv_op_notify_connect_done(struct tevent_req
*subreq
)
80 struct tevent_req
*req
= tevent_req_callback_data(subreq
,
84 status
= dreplsrv_out_drsuapi_recv(subreq
);
86 if (tevent_req_nterror(req
, status
)) {
90 dreplsrv_op_notify_replica_sync_trigger(req
);
93 static void dreplsrv_op_notify_replica_sync_done(struct rpc_request
*rreq
);
95 static void dreplsrv_op_notify_replica_sync_trigger(struct tevent_req
*req
)
97 struct dreplsrv_op_notify_state
*state
=
99 struct dreplsrv_op_notify_state
);
100 struct dreplsrv_partition
*partition
= state
->op
->source_dsa
->partition
;
101 struct dreplsrv_drsuapi_connection
*drsuapi
= state
->op
->source_dsa
->conn
->drsuapi
;
102 struct rpc_request
*rreq
;
103 struct drsuapi_DsReplicaSync
*r
;
105 r
= talloc_zero(state
, struct drsuapi_DsReplicaSync
);
106 if (tevent_req_nomem(r
, req
)) {
109 r
->in
.req
= talloc_zero(r
, union drsuapi_DsReplicaSyncRequest
);
110 if (tevent_req_nomem(r
, req
)) {
113 r
->in
.bind_handle
= &drsuapi
->bind_handle
;
115 r
->in
.req
->req1
.naming_context
= &partition
->nc
;
116 r
->in
.req
->req1
.source_dsa_guid
= state
->op
->service
->ntds_guid
;
117 r
->in
.req
->req1
.options
=
118 DRSUAPI_DRS_ASYNC_OP
|
119 DRSUAPI_DRS_UPDATE_NOTIFICATION
|
120 DRSUAPI_DRS_WRIT_REP
;
122 if (state
->op
->is_urgent
) {
123 r
->in
.req
->req1
.options
|= DRSUAPI_DRS_SYNC_URGENT
;
126 state
->ndr_struct_ptr
= r
;
128 rreq
= dcerpc_drsuapi_DsReplicaSync_send(drsuapi
->pipe
, r
, r
);
129 if (tevent_req_nomem(rreq
, req
)) {
132 composite_continue_rpc(NULL
, rreq
, dreplsrv_op_notify_replica_sync_done
, req
);
135 static void dreplsrv_op_notify_replica_sync_done(struct rpc_request
*rreq
)
137 struct tevent_req
*req
= talloc_get_type(rreq
->async
.private_data
,
139 struct dreplsrv_op_notify_state
*state
=
141 struct dreplsrv_op_notify_state
);
142 struct drsuapi_DsReplicaSync
*r
= talloc_get_type(state
->ndr_struct_ptr
,
143 struct drsuapi_DsReplicaSync
);
146 state
->ndr_struct_ptr
= NULL
;
148 status
= dcerpc_drsuapi_DsReplicaSync_recv(rreq
);
149 if (tevent_req_nterror(req
, status
)) {
153 if (!W_ERROR_IS_OK(r
->out
.result
)) {
154 status
= werror_to_ntstatus(r
->out
.result
);
155 tevent_req_nterror(req
, status
);
159 tevent_req_done(req
);
162 static NTSTATUS
dreplsrv_op_notify_recv(struct tevent_req
*req
)
164 return tevent_req_simple_recv_ntstatus(req
);
167 static void dreplsrv_notify_del_repsTo(struct dreplsrv_notify_operation
*op
)
170 struct repsFromToBlob
*reps
;
172 struct dreplsrv_service
*s
= op
->service
;
175 werr
= dsdb_loadreps(s
->samdb
, op
, op
->source_dsa
->partition
->dn
, "repsTo", &reps
, &count
);
176 if (!W_ERROR_IS_OK(werr
)) {
177 DEBUG(0,(__location__
": Failed to load repsTo for %s\n",
178 ldb_dn_get_linearized(op
->source_dsa
->partition
->dn
)));
182 for (i
=0; i
<count
; i
++) {
183 if (GUID_compare(&reps
[i
].ctr
.ctr1
.source_dsa_obj_guid
,
184 &op
->source_dsa
->repsFrom1
->source_dsa_obj_guid
) == 0) {
185 memmove(&reps
[i
], &reps
[i
+1],
186 sizeof(reps
[i
])*(count
-(i
+1)));
191 werr
= dsdb_savereps(s
->samdb
, op
, op
->source_dsa
->partition
->dn
, "repsTo", reps
, count
);
192 if (!W_ERROR_IS_OK(werr
)) {
193 DEBUG(0,(__location__
": Failed to save repsTo for %s\n",
194 ldb_dn_get_linearized(op
->source_dsa
->partition
->dn
)));
200 called when a notify operation has completed
202 static void dreplsrv_notify_op_callback(struct tevent_req
*subreq
)
204 struct dreplsrv_notify_operation
*op
=
205 tevent_req_callback_data(subreq
,
206 struct dreplsrv_notify_operation
);
208 struct dreplsrv_service
*s
= op
->service
;
210 status
= dreplsrv_op_notify_recv(subreq
);
212 if (!NT_STATUS_IS_OK(status
)) {
213 DEBUG(0,("dreplsrv_notify: Failed to send DsReplicaSync to %s for %s - %s\n",
214 op
->source_dsa
->repsFrom1
->other_info
->dns_name
,
215 ldb_dn_get_linearized(op
->source_dsa
->partition
->dn
),
218 DEBUG(2,("dreplsrv_notify: DsReplicaSync OK for %s\n",
219 op
->source_dsa
->repsFrom1
->other_info
->dns_name
));
220 op
->source_dsa
->notify_uSN
= op
->uSN
;
221 /* delete the repsTo for this replication partner in the
222 partition, as we have successfully told him to sync */
223 dreplsrv_notify_del_repsTo(op
);
227 s
->ops
.n_current
= NULL
;
228 dreplsrv_notify_run_ops(s
);
232 run any pending replica sync calls
234 void dreplsrv_notify_run_ops(struct dreplsrv_service
*s
)
236 struct dreplsrv_notify_operation
*op
;
237 struct tevent_req
*subreq
;
239 if (s
->ops
.n_current
|| s
->ops
.current
) {
240 /* if there's still one running, we're done */
244 if (!s
->ops
.notifies
) {
245 /* if there're no pending operations, we're done */
249 op
= s
->ops
.notifies
;
250 s
->ops
.n_current
= op
;
251 DLIST_REMOVE(s
->ops
.notifies
, op
);
253 subreq
= dreplsrv_op_notify_send(op
, s
->task
->event_ctx
, op
);
255 DEBUG(0,("dreplsrv_notify_run_ops: dreplsrv_op_notify_send[%s][%s] - no memory\n",
256 op
->source_dsa
->repsFrom1
->other_info
->dns_name
,
257 ldb_dn_get_linearized(op
->source_dsa
->partition
->dn
)));
260 tevent_req_set_callback(subreq
, dreplsrv_notify_op_callback
, op
);
265 find a source_dsa for a given guid
267 static struct dreplsrv_partition_source_dsa
*dreplsrv_find_source_dsa(struct dreplsrv_partition
*p
,
270 struct dreplsrv_partition_source_dsa
*s
;
272 for (s
=p
->sources
; s
; s
=s
->next
) {
273 if (GUID_compare(&s
->repsFrom1
->source_dsa_obj_guid
, guid
) == 0) {
282 schedule a replicaSync message
284 static WERROR
dreplsrv_schedule_notify_sync(struct dreplsrv_service
*service
,
285 struct dreplsrv_partition
*p
,
286 struct repsFromToBlob
*reps
,
291 struct dreplsrv_notify_operation
*op
;
292 struct dreplsrv_partition_source_dsa
*s
;
294 s
= dreplsrv_find_source_dsa(p
, &reps
->ctr
.ctr1
.source_dsa_obj_guid
);
296 DEBUG(0,(__location__
": Unable to find source_dsa for %s\n",
297 GUID_string(mem_ctx
, &reps
->ctr
.ctr1
.source_dsa_obj_guid
)));
298 return WERR_DS_UNAVAILABLE
;
301 op
= talloc_zero(mem_ctx
, struct dreplsrv_notify_operation
);
302 W_ERROR_HAVE_NO_MEMORY(op
);
304 op
->service
= service
;
307 op
->is_urgent
= is_urgent
;
309 DLIST_ADD_END(service
->ops
.notifies
, op
, struct dreplsrv_notify_operation
*);
310 talloc_steal(service
, op
);
315 see if a partition has a hugher uSN than what is in the repsTo and
316 if so then send a DsReplicaSync
318 static WERROR
dreplsrv_notify_check(struct dreplsrv_service
*s
,
319 struct dreplsrv_partition
*p
,
323 struct repsFromToBlob
*reps
;
329 werr
= dsdb_loadreps(s
->samdb
, mem_ctx
, p
->dn
, "repsTo", &reps
, &count
);
331 werr
= dsdb_loadreps(s
->samdb
, mem_ctx
, p
->dn
, "repsFrom", &reps
, &count
);
333 if (!W_ERROR_IS_OK(werr
)) {
334 DEBUG(0,(__location__
": Failed to load repsTo for %s\n",
335 ldb_dn_get_linearized(p
->dn
)));
339 /* loads the partition uSNHighest and uSNUrgent */
340 ret
= dsdb_load_partition_usn(s
->samdb
, p
->dn
, &uSNHighest
, &uSNUrgent
);
341 if (ret
!= LDB_SUCCESS
|| uSNHighest
== 0) {
346 /* see if any of our partners need some of our objects */
347 for (i
=0; i
<count
; i
++) {
348 struct dreplsrv_partition_source_dsa
*sdsa
;
349 sdsa
= dreplsrv_find_source_dsa(p
, &reps
[i
].ctr
.ctr1
.source_dsa_obj_guid
);
350 if (sdsa
== NULL
) continue;
351 if (sdsa
->notify_uSN
< uSNHighest
) {
352 /* we need to tell this partner to replicate
355 /* check if urgent replication is needed */
356 if (sdsa
->notify_uSN
< uSNUrgent
) {
357 werr
= dreplsrv_schedule_notify_sync(s
, p
, &reps
[i
], mem_ctx
,
360 werr
= dreplsrv_schedule_notify_sync(s
, p
, &reps
[i
], mem_ctx
,
364 if (!W_ERROR_IS_OK(werr
)) {
365 DEBUG(0,(__location__
": Failed to setup notify to %s for %s\n",
366 reps
[i
].ctr
.ctr1
.other_info
->dns_name
,
367 ldb_dn_get_linearized(p
->dn
)));
377 see if any of the partitions have changed, and if so then send a
378 DsReplicaSync to all the replica partners in the repsTo object
380 static WERROR
dreplsrv_notify_check_all(struct dreplsrv_service
*s
, TALLOC_CTX
*mem_ctx
)
383 struct dreplsrv_partition
*p
;
385 for (p
= s
->partitions
; p
; p
= p
->next
) {
386 status
= dreplsrv_notify_check(s
, p
, mem_ctx
);
387 W_ERROR_NOT_OK_RETURN(status
);
393 static void dreplsrv_notify_run(struct dreplsrv_service
*service
);
395 static void dreplsrv_notify_handler_te(struct tevent_context
*ev
, struct tevent_timer
*te
,
396 struct timeval t
, void *ptr
)
398 struct dreplsrv_service
*service
= talloc_get_type(ptr
, struct dreplsrv_service
);
401 service
->notify
.te
= NULL
;
403 dreplsrv_notify_run(service
);
405 status
= dreplsrv_notify_schedule(service
, service
->notify
.interval
);
406 if (!W_ERROR_IS_OK(status
)) {
407 task_server_terminate(service
->task
, win_errstr(status
), false);
412 WERROR
dreplsrv_notify_schedule(struct dreplsrv_service
*service
, uint32_t next_interval
)
415 struct tevent_timer
*new_te
;
416 struct timeval next_time
;
418 /* prevent looping */
419 if (next_interval
== 0) next_interval
= 1;
421 next_time
= timeval_current_ofs(next_interval
, 50);
423 if (service
->notify
.te
) {
425 * if the timestamp of the new event is higher,
426 * as current next we don't need to reschedule
428 if (timeval_compare(&next_time
, &service
->notify
.next_event
) > 0) {
433 /* reset the next scheduled timestamp */
434 service
->notify
.next_event
= next_time
;
436 new_te
= event_add_timed(service
->task
->event_ctx
, service
,
437 service
->notify
.next_event
,
438 dreplsrv_notify_handler_te
, service
);
439 W_ERROR_HAVE_NO_MEMORY(new_te
);
441 tmp_mem
= talloc_new(service
);
442 DEBUG(4,("dreplsrv_notify_schedule(%u) %sscheduled for: %s\n",
444 (service
->notify
.te
?"re":""),
445 nt_time_string(tmp_mem
, timeval_to_nttime(&next_time
))));
446 talloc_free(tmp_mem
);
448 talloc_free(service
->notify
.te
);
449 service
->notify
.te
= new_te
;
454 static void dreplsrv_notify_run(struct dreplsrv_service
*service
)
458 mem_ctx
= talloc_new(service
);
459 dreplsrv_notify_check_all(service
, mem_ctx
);
460 talloc_free(mem_ctx
);
462 dreplsrv_run_pending_ops(service
);
463 dreplsrv_notify_run_ops(service
);