s4-repl: added a preiodic notification check to the repl task
[ira/wip.git] / source4 / dsdb / repl / drepl_notify.c
1 /* 
2    Unix SMB/CIFS mplementation.
3
4    DSDB replication service periodic notification handling
5    
6    Copyright (C) Andrew Tridgell 2009
7    based on drepl_periodic
8     
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.
13    
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.
18    
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/>.
21    
22 */
23
24 #include "includes.h"
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
38
39 struct dreplsrv_op_notify_state {
40         struct composite_context *creq;
41
42         struct dreplsrv_out_connection *conn;
43
44         struct dreplsrv_drsuapi_connection *drsuapi;
45
46         struct drsuapi_DsBindInfoCtr bind_info_ctr;
47         struct drsuapi_DsBind bind_r;
48         struct dreplsrv_notify_operation *op;
49 };
50
51 /*
52   receive a DsReplicaSync reply
53  */
54 static void dreplsrv_op_notify_replica_sync_recv(struct rpc_request *req)
55 {
56         struct dreplsrv_op_notify_state *st = talloc_get_type(req->async.private_data,
57                                                               struct dreplsrv_op_notify_state);
58         struct composite_context *c = st->creq;
59         struct drsuapi_DsReplicaSync *r = talloc_get_type(req->ndr.struct_ptr,
60                                                           struct drsuapi_DsReplicaSync);
61
62         c->status = dcerpc_ndr_request_recv(req);
63         if (!composite_is_ok(c)) return;
64
65         if (!W_ERROR_IS_OK(r->out.result)) {
66                 composite_error(c, werror_to_ntstatus(r->out.result));
67                 return;
68         }
69
70         composite_done(c);
71 }
72
73 /*
74   send a DsReplicaSync
75 */
76 static void dreplsrv_op_notify_replica_sync_send(struct dreplsrv_op_notify_state *st)
77 {
78         struct composite_context *c = st->creq;
79         struct dreplsrv_partition *partition = st->op->source_dsa->partition;
80         struct dreplsrv_drsuapi_connection *drsuapi = st->op->source_dsa->conn->drsuapi;
81         struct rpc_request *req;
82         struct drsuapi_DsReplicaSync *r;
83
84         r = talloc_zero(st, struct drsuapi_DsReplicaSync);
85         if (composite_nomem(r, c)) return;
86
87         r->in.bind_handle       = &drsuapi->bind_handle;
88         r->in.level = 1;
89         r->in.req.req1.naming_context = &partition->nc;
90         r->in.req.req1.source_dsa_guid = st->op->service->ntds_guid;
91         r->in.req.req1.options = 
92                 DRSUAPI_DS_REPLICA_SYNC_ASYNCHRONOUS_OPERATION |
93                 DRSUAPI_DS_REPLICA_SYNC_WRITEABLE |
94                 DRSUAPI_DS_REPLICA_SYNC_ALL_SOURCES;
95         
96
97         req = dcerpc_drsuapi_DsReplicaSync_send(drsuapi->pipe, r, r);
98         composite_continue_rpc(c, req, dreplsrv_op_notify_replica_sync_recv, st);
99 }
100
101 /*
102   called when we have an established connection
103  */
104 static void dreplsrv_op_notify_connect_recv(struct composite_context *creq)
105 {
106         struct dreplsrv_op_notify_state *st = talloc_get_type(creq->async.private_data,
107                                                               struct dreplsrv_op_notify_state);
108         struct composite_context *c = st->creq;
109
110         c->status = dreplsrv_out_drsuapi_recv(creq);
111         if (!composite_is_ok(c)) return;
112
113         dreplsrv_op_notify_replica_sync_send(st);
114 }
115
116 /*
117   start the ReplicaSync async call
118  */
119 static struct composite_context *dreplsrv_op_notify_send(struct dreplsrv_notify_operation *op)
120 {
121         struct composite_context *c;
122         struct composite_context *creq;
123         struct dreplsrv_op_notify_state *st;
124
125         c = composite_create(op, op->service->task->event_ctx);
126         if (c == NULL) return NULL;
127
128         st = talloc_zero(c, struct dreplsrv_op_notify_state);
129         if (composite_nomem(st, c)) return c;
130
131         st->creq        = c;
132         st->op          = op;
133
134         creq = dreplsrv_out_drsuapi_send(op->source_dsa->conn);
135         composite_continue(c, creq, dreplsrv_op_notify_connect_recv, st);
136
137         return c;
138 }
139
140 static void dreplsrv_notify_del_repsTo(struct dreplsrv_notify_operation *op)
141 {
142         uint32_t count;
143         struct repsFromToBlob *reps;
144         WERROR werr;
145         struct dreplsrv_service *s = op->service;
146         int i;
147
148         werr = dsdb_loadreps(s->samdb, op, op->source_dsa->partition->dn, "repsTo", &reps, &count);
149         if (!W_ERROR_IS_OK(werr)) {
150                 DEBUG(0,(__location__ ": Failed to load repsTo for %s\n",
151                          ldb_dn_get_linearized(op->source_dsa->partition->dn)));
152                 return;
153         }
154
155         for (i=0; i<count; i++) {
156                 if (GUID_compare(&reps[i].ctr.ctr1.source_dsa_obj_guid, 
157                                  &op->source_dsa->repsFrom1->source_dsa_obj_guid) == 0) {
158                         memmove(&reps[i], &reps[i+1],
159                                 sizeof(reps[i])*(count-(i+1)));
160                         count--;
161                 }
162         }
163
164         werr = dsdb_savereps(s->samdb, op, op->source_dsa->partition->dn, "repsTo", reps, count);
165         if (!W_ERROR_IS_OK(werr)) {
166                 DEBUG(0,(__location__ ": Failed to save repsTo for %s\n",
167                          ldb_dn_get_linearized(op->source_dsa->partition->dn)));
168                 return;
169         }
170 }
171
172 /*
173   called when a notify operation has completed
174  */
175 static void dreplsrv_notify_op_callback(struct dreplsrv_notify_operation *op)
176 {
177         NTSTATUS status;
178         struct dreplsrv_service *s = op->service;
179
180         status = composite_wait(op->creq);
181         if (!NT_STATUS_IS_OK(status)) {
182                 DEBUG(0,("dreplsrv_notify: Failed to send DsReplicaSync to %s - %s\n",
183                          op->source_dsa->repsFrom1->other_info->dns_name,
184                          nt_errstr(status)));
185         } else {
186                 DEBUG(2,("dreplsrv_notify: DsReplicaSync OK for %s\n",
187                          op->source_dsa->repsFrom1->other_info->dns_name));
188                 op->source_dsa->notify_uSN = op->uSN;
189                 /* delete the repsTo for this replication partner in the
190                    partition, as we have successfully told him to sync */
191                 dreplsrv_notify_del_repsTo(op);
192         }
193         talloc_free(op->creq);
194
195         talloc_free(op);
196         s->ops.n_current = NULL;
197         dreplsrv_notify_run_ops(s);
198 }
199
200
201 static void dreplsrv_notify_op_callback_creq(struct composite_context *creq)
202 {
203         struct dreplsrv_notify_operation *op = talloc_get_type(creq->async.private_data,
204                                                                struct dreplsrv_notify_operation);
205         dreplsrv_notify_op_callback(op);
206 }
207
208 /*
209   run any pending replica sync calls
210  */
211 void dreplsrv_notify_run_ops(struct dreplsrv_service *s)
212 {
213         struct dreplsrv_notify_operation *op;
214
215         if (s->ops.n_current || s->ops.current) {
216                 /* if there's still one running, we're done */
217                 return;
218         }
219
220         if (!s->ops.notifies) {
221                 /* if there're no pending operations, we're done */
222                 return;
223         }
224
225         op = s->ops.notifies;
226         s->ops.n_current = op;
227         DLIST_REMOVE(s->ops.notifies, op);
228
229         op->creq = dreplsrv_op_notify_send(op);
230         if (!op->creq) {
231                 dreplsrv_notify_op_callback(op);
232                 return;
233         }
234
235         op->creq->async.fn              = dreplsrv_notify_op_callback_creq;
236         op->creq->async.private_data    = op;
237 }
238
239
240 /*
241   find a source_dsa for a given guid
242  */
243 static struct dreplsrv_partition_source_dsa *dreplsrv_find_source_dsa(struct dreplsrv_partition *p,
244                                                                       struct GUID *guid)
245 {
246         struct dreplsrv_partition_source_dsa *s;
247
248         for (s=p->sources; s; s=s->next) {
249                 if (GUID_compare(&s->repsFrom1->source_dsa_obj_guid, guid) == 0) {
250                         return s;
251                 }
252         }
253         return NULL;
254 }
255
256
257 /*
258   schedule a replicaSync message
259  */
260 static WERROR dreplsrv_schedule_notify_sync(struct dreplsrv_service *service,
261                                             struct dreplsrv_partition *p,
262                                             struct repsFromToBlob *reps,
263                                             TALLOC_CTX *mem_ctx,
264                                             uint64_t uSN)
265 {
266         struct dreplsrv_notify_operation *op;
267         struct dreplsrv_partition_source_dsa *s;
268
269         s = dreplsrv_find_source_dsa(p, &reps->ctr.ctr1.source_dsa_obj_guid);
270         if (s == NULL) {
271                 DEBUG(0,(__location__ ": Unable to find source_dsa for %s\n",
272                          GUID_string(mem_ctx, &reps->ctr.ctr1.source_dsa_obj_guid)));
273                 return WERR_DS_UNAVAILABLE;
274         }
275
276         op = talloc_zero(mem_ctx, struct dreplsrv_notify_operation);
277         W_ERROR_HAVE_NO_MEMORY(op);
278
279         op->service     = service;
280         op->source_dsa  = s;
281         op->uSN         = uSN;
282
283         DLIST_ADD_END(service->ops.notifies, op, struct dreplsrv_notify_operation *);
284         talloc_steal(service, op);
285         return WERR_OK;
286 }
287
288 /*
289   see if a partition has a hugher uSN than what is in the repsTo and
290   if so then send a DsReplicaSync
291  */
292 static WERROR dreplsrv_notify_check(struct dreplsrv_service *s, 
293                                     struct dreplsrv_partition *p,
294                                     TALLOC_CTX *mem_ctx)
295 {
296         uint32_t count;
297         struct repsFromToBlob *reps;
298         WERROR werr;
299         uint64_t uSN;
300         int ret, i;
301
302         werr = dsdb_loadreps(s->samdb, mem_ctx, p->dn, "repsTo", &reps, &count);
303         if (!W_ERROR_IS_OK(werr)) {
304                 DEBUG(0,(__location__ ": Failed to load repsTo for %s\n",
305                          ldb_dn_get_linearized(p->dn)));
306                 return werr;
307         }
308
309         /* loads the partition uSNHighest */
310         ret = dsdb_load_partition_usn(s->samdb, p->dn, &uSN);
311         if (ret != LDB_SUCCESS || uSN == 0) {
312                 /* nothing to do */
313                 return WERR_OK;
314         }
315
316         /* see if any of our partners need some of our objects */
317         for (i=0; i<count; i++) {
318                 struct dreplsrv_partition_source_dsa *sdsa;
319                 sdsa = dreplsrv_find_source_dsa(p, &reps[i].ctr.ctr1.source_dsa_obj_guid);
320                 if (sdsa == NULL) continue;
321                 if (sdsa->notify_uSN < uSN) {
322                         /* we need to tell this partner to replicate
323                            with us */
324                         werr = dreplsrv_schedule_notify_sync(s, p, &reps[i], mem_ctx, uSN);
325                         if (!W_ERROR_IS_OK(werr)) {
326                                 DEBUG(0,(__location__ ": Failed to setup notify to %s for %s\n",
327                                          reps[i].ctr.ctr1.other_info->dns_name,
328                                          ldb_dn_get_linearized(p->dn)));
329                                 return werr;
330                         }
331                 }
332         }
333
334
335         return WERR_OK;
336 }
337
338 /*
339   see if any of the partitions have changed, and if so then send a
340   DsReplicaSync to all the replica partners in the repsTo object
341  */
342 static WERROR dreplsrv_notify_check_all(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx)
343 {
344         WERROR status;
345         struct dreplsrv_partition *p;
346
347         for (p = s->partitions; p; p = p->next) {
348                 status = dreplsrv_notify_check(s, p, mem_ctx);
349                 W_ERROR_NOT_OK_RETURN(status);
350         }
351
352         return WERR_OK;
353 }
354
355 static void dreplsrv_notify_run(struct dreplsrv_service *service);
356
357 static void dreplsrv_notify_handler_te(struct tevent_context *ev, struct tevent_timer *te,
358                                        struct timeval t, void *ptr)
359 {
360         struct dreplsrv_service *service = talloc_get_type(ptr, struct dreplsrv_service);
361         WERROR status;
362
363         service->notify.te = NULL;
364
365         dreplsrv_notify_run(service);
366
367         status = dreplsrv_notify_schedule(service, service->notify.interval);
368         if (!W_ERROR_IS_OK(status)) {
369                 task_server_terminate(service->task, win_errstr(status));
370                 return;
371         }
372 }
373
374 WERROR dreplsrv_notify_schedule(struct dreplsrv_service *service, uint32_t next_interval)
375 {
376         TALLOC_CTX *tmp_mem;
377         struct tevent_timer *new_te;
378         struct timeval next_time;
379
380         /* prevent looping */
381         if (next_interval == 0) next_interval = 1;
382
383         next_time = timeval_current_ofs(next_interval, 50);
384
385         if (service->notify.te) {
386                 /*
387                  * if the timestamp of the new event is higher,
388                  * as current next we don't need to reschedule
389                  */
390                 if (timeval_compare(&next_time, &service->notify.next_event) > 0) {
391                         return WERR_OK;
392                 }
393         }
394
395         /* reset the next scheduled timestamp */
396         service->notify.next_event = next_time;
397
398         new_te = event_add_timed(service->task->event_ctx, service,
399                                  service->notify.next_event,
400                                  dreplsrv_notify_handler_te, service);
401         W_ERROR_HAVE_NO_MEMORY(new_te);
402
403         tmp_mem = talloc_new(service);
404         DEBUG(2,("dreplsrv_notify_schedule(%u) %sscheduled for: %s\n",
405                 next_interval,
406                 (service->notify.te?"re":""),
407                 nt_time_string(tmp_mem, timeval_to_nttime(&next_time))));
408         talloc_free(tmp_mem);
409
410         talloc_free(service->notify.te);
411         service->notify.te = new_te;
412
413         return WERR_OK;
414 }
415
416 static void dreplsrv_notify_run(struct dreplsrv_service *service)
417 {
418         TALLOC_CTX *mem_ctx;
419
420         mem_ctx = talloc_new(service);
421         dreplsrv_notify_check_all(service, mem_ctx);
422         talloc_free(mem_ctx);
423
424         dreplsrv_run_pending_ops(service);
425         dreplsrv_notify_run_ops(service);
426 }