s4-repl: added a preiodic notification check to the repl task
[ira/wip.git] / source4 / dsdb / repl / drepl_out_pull.c
1 /* 
2    Unix SMB/CIFS mplementation.
3    DSDB replication service outgoing Pull-Replication
4    
5    Copyright (C) Stefan Metzmacher 2007
6     
7    This program is free software; you can redistribute it and/or modify
8    it under the terms of the GNU General Public License as published by
9    the Free Software Foundation; either version 3 of the License, or
10    (at your option) any later version.
11    
12    This program is distributed in the hope that it will be useful,
13    but WITHOUT ANY WARRANTY; without even the implied warranty of
14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15    GNU General Public License for more details.
16    
17    You should have received a copy of the GNU General Public License
18    along with this program.  If not, see <http://www.gnu.org/licenses/>.
19    
20 */
21
22 #include "includes.h"
23 #include "dsdb/samdb/samdb.h"
24 #include "auth/auth.h"
25 #include "smbd/service.h"
26 #include "lib/events/events.h"
27 #include "lib/messaging/irpc.h"
28 #include "dsdb/repl/drepl_service.h"
29 #include "lib/ldb/include/ldb_errors.h"
30 #include "../lib/util/dlinklist.h"
31 #include "librpc/gen_ndr/ndr_misc.h"
32 #include "librpc/gen_ndr/ndr_drsuapi.h"
33 #include "librpc/gen_ndr/ndr_drsblobs.h"
34 #include "libcli/composite/composite.h"
35
36 static WERROR dreplsrv_schedule_partition_pull_source(struct dreplsrv_service *s,
37                                                       struct dreplsrv_partition *p,
38                                                       struct dreplsrv_partition_source_dsa *source,
39                                                       TALLOC_CTX *mem_ctx)
40 {
41         struct dreplsrv_out_operation *op;
42
43         op = talloc_zero(mem_ctx, struct dreplsrv_out_operation);
44         W_ERROR_HAVE_NO_MEMORY(op);
45
46         op->service     = s;
47         op->source_dsa  = source;
48
49         DLIST_ADD_END(s->ops.pending, op, struct dreplsrv_out_operation *);
50         talloc_steal(s, op);
51         return WERR_OK;
52 }
53
54 static WERROR dreplsrv_schedule_partition_pull(struct dreplsrv_service *s,
55                                                struct dreplsrv_partition *p,
56                                                TALLOC_CTX *mem_ctx)
57 {
58         WERROR status;
59         struct dreplsrv_partition_source_dsa *cur;
60
61         for (cur = p->sources; cur; cur = cur->next) {
62                 status = dreplsrv_schedule_partition_pull_source(s, p, cur, mem_ctx);
63                 W_ERROR_NOT_OK_RETURN(status);
64         }
65
66         return WERR_OK;
67 }
68
69 WERROR dreplsrv_schedule_pull_replication(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx)
70 {
71         WERROR status;
72         struct dreplsrv_partition *p;
73
74         for (p = s->partitions; p; p = p->next) {
75                 status = dreplsrv_schedule_partition_pull(s, p, mem_ctx);
76                 W_ERROR_NOT_OK_RETURN(status);
77         }
78
79         return WERR_OK;
80 }
81
82
83 /* force an immediate of the specified partition by GUID  */
84 WERROR dreplsrv_schedule_partition_pull_by_guid(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx,
85                                                 struct GUID *guid)
86 {
87         struct dreplsrv_partition *p;
88         
89         for (p = s->partitions; p; p = p->next) {
90                 if (GUID_compare(&p->nc.guid, guid) == 0) {
91                         return dreplsrv_schedule_partition_pull(s, p, mem_ctx);
92                 }
93         }
94
95         return WERR_NOT_FOUND;
96 }
97
98 static void dreplsrv_pending_op_callback(struct dreplsrv_out_operation *op)
99 {
100         struct repsFromTo1 *rf = op->source_dsa->repsFrom1;
101         struct dreplsrv_service *s = op->service;
102         time_t t;
103         NTTIME now;
104
105         t = time(NULL);
106         unix_to_nt_time(&now, t);
107
108         rf->result_last_attempt = dreplsrv_op_pull_source_recv(op->creq);
109         if (W_ERROR_IS_OK(rf->result_last_attempt)) {
110                 rf->consecutive_sync_failures   = 0;
111                 rf->last_success                = now;
112                 DEBUG(2,("dreplsrv_op_pull_source(%s)\n",
113                         win_errstr(rf->result_last_attempt)));
114                 goto done;
115         }
116
117         rf->consecutive_sync_failures++;
118
119         DEBUG(1,("dreplsrv_op_pull_source(%s/%s) failures[%u]\n",
120                 win_errstr(rf->result_last_attempt),
121                 nt_errstr(werror_to_ntstatus(rf->result_last_attempt)),
122                 rf->consecutive_sync_failures));
123
124 done:
125         talloc_free(op);
126         s->ops.current = NULL;
127         dreplsrv_run_pending_ops(s);
128         dreplsrv_notify_run_ops(s);
129 }
130
131 static void dreplsrv_pending_op_callback_creq(struct composite_context *creq)
132 {
133         struct dreplsrv_out_operation *op = talloc_get_type(creq->async.private_data,
134                                                            struct dreplsrv_out_operation);
135         dreplsrv_pending_op_callback(op);
136 }
137
138 void dreplsrv_run_pending_ops(struct dreplsrv_service *s)
139 {
140         struct dreplsrv_out_operation *op;
141         time_t t;
142         NTTIME now;
143
144         if (s->ops.current || s->ops.n_current) {
145                 /* if there's still one running, we're done */
146                 return;
147         }
148
149         if (!s->ops.pending) {
150                 /* if there're no pending operations, we're done */
151                 return;
152         }
153
154         t = time(NULL);
155         unix_to_nt_time(&now, t);
156
157         op = s->ops.pending;
158         s->ops.current = op;
159         DLIST_REMOVE(s->ops.pending, op);
160
161         op->source_dsa->repsFrom1->last_attempt = now;
162
163         op->creq = dreplsrv_op_pull_source_send(op);
164         if (!op->creq) {
165                 dreplsrv_pending_op_callback(op);
166                 return;
167         }
168
169         op->creq->async.fn              = dreplsrv_pending_op_callback_creq;
170         op->creq->async.private_data    = op;
171 }