s4:drepl_out_helpers: don't look at the internals of 'struct rpc_request'
[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 WERROR dreplsrv_schedule_partition_pull_source(struct dreplsrv_service *s,
37                                                struct dreplsrv_partition_source_dsa *source,
38                                                enum drsuapi_DsExtendedOperation extended_op,
39                                                uint64_t fsmo_info,
40                                                dreplsrv_fsmo_callback_t callback)
41 {
42         struct dreplsrv_out_operation *op;
43
44         op = talloc_zero(s, struct dreplsrv_out_operation);
45         W_ERROR_HAVE_NO_MEMORY(op);
46
47         op->service     = s;
48         op->source_dsa  = source;
49         op->extended_op = extended_op;
50         op->fsmo_info   = fsmo_info;
51         op->callback    = callback;
52
53         DLIST_ADD_END(s->ops.pending, op, struct dreplsrv_out_operation *);
54
55         return WERR_OK;
56 }
57
58 static WERROR dreplsrv_schedule_partition_pull(struct dreplsrv_service *s,
59                                                struct dreplsrv_partition *p,
60                                                TALLOC_CTX *mem_ctx)
61 {
62         WERROR status;
63         struct dreplsrv_partition_source_dsa *cur;
64
65         for (cur = p->sources; cur; cur = cur->next) {
66                 status = dreplsrv_schedule_partition_pull_source(s, cur, DRSUAPI_EXOP_NONE, 0, NULL);
67                 W_ERROR_NOT_OK_RETURN(status);
68         }
69
70         return WERR_OK;
71 }
72
73 WERROR dreplsrv_schedule_pull_replication(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx)
74 {
75         WERROR status;
76         struct dreplsrv_partition *p;
77
78         for (p = s->partitions; p; p = p->next) {
79                 status = dreplsrv_schedule_partition_pull(s, p, mem_ctx);
80                 W_ERROR_NOT_OK_RETURN(status);
81         }
82
83         return WERR_OK;
84 }
85
86
87 /* force an immediate of the specified partition by GUID  */
88 WERROR dreplsrv_schedule_partition_pull_by_guid(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx,
89                                                 struct GUID *guid)
90 {
91         struct dreplsrv_partition *p;
92         
93         for (p = s->partitions; p; p = p->next) {
94                 if (GUID_compare(&p->nc.guid, guid) == 0) {
95                         return dreplsrv_schedule_partition_pull(s, p, mem_ctx);
96                 }
97         }
98
99         return WERR_NOT_FOUND;
100 }
101
102 static void dreplsrv_pending_op_callback(struct tevent_req *subreq)
103 {
104         struct dreplsrv_out_operation *op = tevent_req_callback_data(subreq,
105                                             struct dreplsrv_out_operation);
106         struct repsFromTo1 *rf = op->source_dsa->repsFrom1;
107         struct dreplsrv_service *s = op->service;
108         time_t t;
109         NTTIME now;
110
111         t = time(NULL);
112         unix_to_nt_time(&now, t);
113
114         rf->result_last_attempt = dreplsrv_op_pull_source_recv(subreq);
115         TALLOC_FREE(subreq);
116         if (W_ERROR_IS_OK(rf->result_last_attempt)) {
117                 rf->consecutive_sync_failures   = 0;
118                 rf->last_success                = now;
119                 DEBUG(3,("dreplsrv_op_pull_source(%s)\n",
120                         win_errstr(rf->result_last_attempt)));
121                 goto done;
122         }
123
124         rf->consecutive_sync_failures++;
125
126         DEBUG(1,("dreplsrv_op_pull_source(%s/%s) for %s failures[%u]\n",
127                  win_errstr(rf->result_last_attempt),
128                  win_errstr(rf->result_last_attempt),
129                  ldb_dn_get_linearized(op->source_dsa->partition->dn),
130                  rf->consecutive_sync_failures));
131
132 done:
133         if (op->callback) {
134                 op->callback(s, rf->result_last_attempt);
135         }
136         talloc_free(op);
137         s->ops.current = NULL;
138         dreplsrv_run_pending_ops(s);
139         dreplsrv_notify_run_ops(s);
140 }
141
142 void dreplsrv_run_pending_ops(struct dreplsrv_service *s)
143 {
144         struct dreplsrv_out_operation *op;
145         time_t t;
146         NTTIME now;
147         struct tevent_req *subreq;
148
149         if (s->ops.current || s->ops.n_current) {
150                 /* if there's still one running, we're done */
151                 return;
152         }
153
154         if (!s->ops.pending) {
155                 /* if there're no pending operations, we're done */
156                 return;
157         }
158
159         t = time(NULL);
160         unix_to_nt_time(&now, t);
161
162         op = s->ops.pending;
163         s->ops.current = op;
164         DLIST_REMOVE(s->ops.pending, op);
165
166         op->source_dsa->repsFrom1->last_attempt = now;
167
168         subreq = dreplsrv_op_pull_source_send(op, s->task->event_ctx, op);
169         if (!subreq) {
170                 struct repsFromTo1 *rf = op->source_dsa->repsFrom1;
171
172                 rf->result_last_attempt = WERR_NOMEM;
173                 rf->consecutive_sync_failures++;
174
175                 DEBUG(1,("dreplsrv_op_pull_source(%s/%s) failures[%u]\n",
176                         win_errstr(rf->result_last_attempt),
177                         nt_errstr(werror_to_ntstatus(rf->result_last_attempt)),
178                         rf->consecutive_sync_failures));
179                 return;
180         }
181         tevent_req_set_callback(subreq, dreplsrv_pending_op_callback, op);
182 }